orchestrata_ultimate.postman.guarantees

Reference › orchestrata_ultimate › postman › guarantees

The delivery-guarantee bookkeeping of the broker, SPEC-002 story US-007 (REQ-002-US-007-001 to -004), with retry and timeout requirements traced to SPEC-001 story US-002. DeliveryGuaranteeManager implements IDeliveryGuarantee from interfaces and tracks acknowledgements, retries and idempotency for the three DeliveryMode values defined in types. Defaults (3 retries, a one-second delay doubling per attempt, a 30-second acknowledgement timeout) come from constants. It tracks state and does not itself transport messages; no other module imports it yet.

send(message_id, destinations, mode) records the mode and, depending on it, does one of these: for at-most-once it returns a queued result with received receipts and tracks nothing more; for exactly-once it builds an idempotency token message_id:destination for each destination and drops destinations already seen, returning a delivered result with no destinations when none remain; for at-least-once, and for the destinations kept in exactly-once mode, it records the set of destinations still owing an acknowledgement, the send time per destination and a retry counter of zero. acknowledge removes a destination from the pending set and cleans up the message when none remain; negative_acknowledge records a failure; get_pending_acks lists who has not acknowledged. retry returns False when the retry limit is reached, otherwise waits the backoff delay (capped at 60 seconds), increments the counter and returns True. is_duplicate checks the idempotency set. check_timeouts returns the message and destination pairs whose acknowledgement is older than the timeout, and drops them from the pending set for at-most-once messages. get_stats reports a message's counters. A lock guards all tables.

At a glance

Class diagram

Dependency diagram

API

Classes

DeliveryGuaranteeManager

Bases: IDeliveryGuarantee · Decorators: traces(spec='002', story='US-007', reqs=['REQ-002-US-007-001', 'REQ-002-US-007-002', 'REQ-002-US-007-003']) · Line 53

Manages delivery guarantees for Postman message broker.

Source

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

# mypy: ignore-errors
"""
Delivery Guarantee System for Postman Message Broker.

Implements IDeliveryGuarantee interface with support for:
- AT_MOST_ONCE: Fire and forget
- AT_LEAST_ONCE: Retry until delivered
- EXACTLY_ONCE: Single delivery with deduplication

Features:
- ACK/NACK tracking per message per destination
- Exponential backoff retry logic
- Timeout handling for unacknowledged messages
- Idempotency tokens for exactly-once delivery
- Thread-safe operation with asyncio
"""

import asyncio
import logging
from datetime import UTC, datetime, timedelta
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.constants import (
    DEFAULT_ACK_TIMEOUT_SECONDS,
    DEFAULT_MAX_RETRIES,
    DEFAULT_RETRY_BACKOFF_MULTIPLIER,
    DEFAULT_RETRY_DELAY_MS,
    MAX_RETRY_DELAY_MS,
)
from orchestrata_ultimate.postman.exceptions import MaxRetriesExceededError
from orchestrata_ultimate.postman.interfaces import IDeliveryGuarantee
from orchestrata_ultimate.postman.types import (
    DeliveryMode,
    DeliveryReceipt,
    DeliveryResult,
    DeliveryStatus,
    ReceiptType,
)

logger = logging.getLogger(__name__)


@traces(
    spec="002",
    story="US-007",
    reqs=[
        "REQ-002-US-007-001",
        "REQ-002-US-007-002",
        "REQ-002-US-007-003",
    ],
)
class DeliveryGuaranteeManager(IDeliveryGuarantee):
    """
    Manages delivery guarantees for Postman message broker.

    Tracks acknowledgments, handles retries with exponential backoff,
    and ensures delivery semantics based on configured mode.
    """

    def __init__(
        self,
        max_retries: int = DEFAULT_MAX_RETRIES,
        retry_delay_ms: int = DEFAULT_RETRY_DELAY_MS,
        retry_backoff: float = DEFAULT_RETRY_BACKOFF_MULTIPLIER,
        ack_timeout_seconds: int = DEFAULT_ACK_TIMEOUT_SECONDS,
    ) -> None:
        """
        Initialize delivery guarantee manager.

        Args:
            max_retries: Maximum retry attempts per delivery
            retry_delay_ms: Initial retry delay in milliseconds
            retry_backoff: Exponential backoff multiplier
            ack_timeout_seconds: Timeout for acknowledgments
        """
        self._max_retries = max_retries
        self._retry_delay_ms = retry_delay_ms
        self._retry_backoff = retry_backoff
        self._ack_timeout_seconds = ack_timeout_seconds

        # Tracking structures
        self._pending_acks: dict[str, set[str]] = {}  # message_id -> destinations
        self._ack_times: dict[str, dict[str, datetime]] = (
            {}
        )  # message_id -> dest -> time
        self._retries: dict[str, dict[str, int]] = {}  # message_id -> dest -> count
        self._idempotency: set[str] = set()  # "message_id:destination" tokens
        self._delivery_modes: dict[str, DeliveryMode] = {}  # message_id -> mode

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

    @traces(spec="002", story="US-007", reqs=["REQ-002-US-007-004"])
    async def send(
        self,
        message_id: str,
        destinations: list[str],
        mode: DeliveryMode,
    ) -> DeliveryResult:
        """
        Send message with delivery guarantee.

        Args:
            message_id: Unique message identifier
            destinations: List of destination addresses
            mode: Delivery guarantee mode

        Returns:
            DeliveryResult with status and receipts

        Raises:
            DeliveryError: If delivery fails based on mode requirements
        """
        async with self._lock:
            # Store delivery mode
            self._delivery_modes[message_id] = mode

            # Handle AT_MOST_ONCE (fire and forget)
            if mode == DeliveryMode.AT_MOST_ONCE:
                return DeliveryResult(
                    success=True,
                    message_id=message_id,
                    destinations=destinations,
                    status=DeliveryStatus.QUEUED,
                    receipts=[
                        DeliveryReceipt(
                            message_id=message_id,
                            recipient=dest,
                            receipt_type=ReceiptType.RECEIVED,
                            timestamp=datetime.now(UTC),
                        )
                        for dest in destinations
                    ],
                )

            # Handle EXACTLY_ONCE (check for duplicates)
            if mode == DeliveryMode.EXACTLY_ONCE:
                filtered_destinations = []
                for dest in destinations:
                    token = f"{message_id}:{dest}"
                    if token not in self._idempotency:
                        filtered_destinations.append(dest)
                        self._idempotency.add(token)
                destinations = filtered_destinations

                if not destinations:
                    # All destinations already received this message
                    return DeliveryResult(
                        success=True,
                        message_id=message_id,
                        destinations=[],
                        status=DeliveryStatus.DELIVERED,
                    )

            # Initialize tracking for AT_LEAST_ONCE and EXACTLY_ONCE
            self._pending_acks[message_id] = set(destinations)
            self._ack_times[message_id] = {
                dest: datetime.now(UTC) for dest in destinations
            }
            self._retries[message_id] = {dest: 0 for dest in destinations}

            return DeliveryResult(
                success=True,
                message_id=message_id,
                destinations=destinations,
                status=DeliveryStatus.QUEUED,
                receipts=[
                    DeliveryReceipt(
                        message_id=message_id,
                        recipient=dest,
                        receipt_type=ReceiptType.RECEIVED,
                        timestamp=datetime.now(UTC),
                    )
                    for dest in destinations
                ],
            )

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-034"])
    async def acknowledge(self, message_id: str, destination: str) -> None:
        """
        Record acknowledgment from destination.

        Args:
            message_id: Message identifier
            destination: Destination that acknowledged
        """
        async with self._lock:
            if message_id in self._pending_acks:
                self._pending_acks[message_id].discard(destination)

                # Clean up if all destinations acknowledged
                if not self._pending_acks[message_id]:
                    self._cleanup_message(message_id)

    @traces(spec="002", story="US-007", reqs=["REQ-002-US-007-008"])
    async def negative_acknowledge(
        self, message_id: str, destination: str, reason: str
    ) -> None:
        """
        Record negative acknowledgment (delivery failure).

        Args:
            message_id: Message identifier
            destination: Destination that NACK'd
            reason: Failure reason

        Raises:
            MaxRetriesExceededError: If max retries exceeded for this destination
        """
        async with self._lock:
            if message_id not in self._retries:
                return

            mode = self._delivery_modes.get(message_id, DeliveryMode.AT_MOST_ONCE)

            # AT_MOST_ONCE doesn't retry
            if mode == DeliveryMode.AT_MOST_ONCE:
                self._pending_acks[message_id].discard(destination)
                return

            # Increment retry count
            current_retries = self._retries[message_id].get(destination, 0)
            self._retries[message_id][destination] = current_retries + 1

            # Check if max retries exceeded
            if current_retries >= self._max_retries:
                self._pending_acks[message_id].discard(destination)

                # Clean up if no more pending
                if not self._pending_acks[message_id]:
                    self._cleanup_message(message_id)

                raise MaxRetriesExceededError(
                    f"Max retries ({self._max_retries}) exceeded for "
                    f"message {message_id} to {destination}: {reason}"
                )

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-036"])
    async def get_pending_acks(self, message_id: str) -> list[str]:
        """
        Get list of destinations with pending acknowledgments.

        Args:
            message_id: Message identifier

        Returns:
            List of destination addresses awaiting ACK
        """
        async with self._lock:
            return list(self._pending_acks.get(message_id, set()))

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-015"])
    async def retry(self, message_id: str, destination: str) -> bool:
        """
        Retry delivery to specific destination.

        Args:
            message_id: Message identifier
            destination: Destination to retry

        Returns:
            True if retry scheduled, False if max retries exceeded
        """
        async with self._lock:
            if message_id not in self._retries:
                return False

            current_retries = self._retries[message_id].get(destination, 0)
            if current_retries >= self._max_retries:
                return False

            # Calculate backoff delay
            delay_ms = min(
                self._retry_delay_ms * (self._retry_backoff**current_retries),
                MAX_RETRY_DELAY_MS,
            )

            # Update retry count
            self._retries[message_id][destination] = current_retries + 1

            # Schedule retry (caller handles actual retry logic)
            await asyncio.sleep(delay_ms / 1000.0)

            return True

    @traces(spec="002", story="US-007", reqs=["REQ-002-US-007-003"])
    async def is_duplicate(self, message_id: str, destination: str) -> bool:
        """
        Check if message is duplicate for exactly-once delivery.

        Args:
            message_id: Message identifier
            destination: Destination address

        Returns:
            True if duplicate, False if first delivery
        """
        async with self._lock:
            token = f"{message_id}:{destination}"
            return token in self._idempotency

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-017"])
    async def check_timeouts(self) -> list[tuple[str, str]]:
        """
        Check for timed-out acknowledgments.

        Returns:
            List of (message_id, destination) tuples that timed out
        """
        async with self._lock:
            now = datetime.now(UTC)
            timeout_delta = timedelta(seconds=self._ack_timeout_seconds)
            timed_out = []

            for message_id, destinations in list(self._ack_times.items()):
                for destination, ack_time in list(destinations.items()):
                    if now - ack_time > timeout_delta:
                        timed_out.append((message_id, destination))

                        # Remove from pending if AT_MOST_ONCE
                        mode = self._delivery_modes.get(
                            message_id, DeliveryMode.AT_MOST_ONCE
                        )
                        if mode == DeliveryMode.AT_MOST_ONCE:
                            self._pending_acks[message_id].discard(destination)

            return timed_out

    def _cleanup_message(self, message_id: str) -> None:
        """
        Clean up tracking data for completed message.

        Args:
            message_id: Message identifier to clean up
        """
        self._pending_acks.pop(message_id, None)
        self._ack_times.pop(message_id, None)
        self._retries.pop(message_id, None)
        self._delivery_modes.pop(message_id, None)

        # Note: Keep idempotency tokens indefinitely for EXACTLY_ONCE
        # In production, would need TTL or periodic cleanup

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-036"])
    async def get_stats(self, message_id: str) -> dict[str, Any]:
        """
        Get delivery statistics for message.

        Args:
            message_id: Message identifier

        Returns:
            Dictionary with pending count, retry counts, mode
        """
        async with self._lock:
            return {
                "pending": len(self._pending_acks.get(message_id, set())),
                "retries": dict(self._retries.get(message_id, {})),
                "mode": self._delivery_modes.get(
                    message_id, DeliveryMode.AT_MOST_ONCE
                ).value,
            }