orchestrata_ultimate.postman.error_handler

Reference › orchestrata_ultimate › postman › error_handler

Retry and circuit-breaker policy for the broker, SPEC-002 story FR-008 (REQ-002-FR8-016 to -020) and REQ-002-US-007-014. PostmanErrorHandler implements IErrorHandler from interfaces and decides whether a failed delivery should be retried. Defaults come from constants, and the error classes from exceptions. Nothing in the package imports it yet.

CircuitBreaker is a per-destination breaker. record_failure counts failures and sets the state to open once the failure threshold (5) is reached; is_open reports True while open and moves to half-open once the reset timeout (60 seconds) has passed; record_success clears the count. PostmanErrorHandler keeps one breaker per destination. handle(error, context) reads the destination and the retry attempt from the context and returns a pair of retry flag and message. It refuses when the circuit is open, when the attempt count has reached max_retries (3 by default, and records a failure), when the error is a CircuitOpenError or MaxRetriesExceededError, or when it is a DeliveryFailedError (a permanent failure). Otherwise it records a failure and allows a retry after the delay from _calculate_backoff: a base delay of 100 ms doubled per attempt, with plus or minus 10% random jitter. is_circuit_open, record_success and record_failure expose the breaker state.

At a glance

Class diagram

Dependency diagram

API

Classes

CircuitBreaker

Decorators: traces(spec='002', story='FR-008', reqs=['REQ-002-FR8-020']) · Line 33

Simple circuit breaker per destination.

PostmanErrorHandler

Bases: IErrorHandler · Decorators: traces(spec='002', story='FR-008', reqs=['REQ-002-FR8-016', 'REQ-002-FR8-017', 'REQ-002-FR8-019', 'REQ-002-FR8-020', 'R… · Line 85

Error handler with circuit breakers and exponential backoff.

Source

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

# mypy: ignore-errors
"""Error handling with circuit breaker and retry backoff for Postman.

Implements IErrorHandler interface.
"""

import asyncio
import logging
import random
import time
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.constants import (
    CIRCUIT_BREAKER_FAILURE_THRESHOLD,
    CIRCUIT_BREAKER_RESET_TIMEOUT_SECONDS,
    RETRY_BASE_DELAY_MS,
)
from orchestrata_ultimate.postman.exceptions import (
    CircuitOpenError,
    DeliveryFailedError,
    MaxRetriesExceededError,
)
from orchestrata_ultimate.postman.interfaces import IErrorHandler

# Spec: 002 - Postman Message Broker


logger = logging.getLogger(__name__)


@traces(spec="002", story="FR-008", reqs=["REQ-002-FR8-020"])
class CircuitBreaker:
    """Simple circuit breaker per destination."""

    @traces(spec="002", story="FR-008", reqs=["REQ-002-FR8-020"])
    def __init__(
        self,
        failure_threshold: int = CIRCUIT_BREAKER_FAILURE_THRESHOLD,
        reset_timeout: float = float(CIRCUIT_BREAKER_RESET_TIMEOUT_SECONDS),
    ) -> None:
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.last_failure: float | None = None
        self.state = "closed"  # closed, open, half_open

    @traces(spec="002", story="FR-008", reqs=["REQ-002-FR8-020"])
    def is_open(self) -> bool:
        if self.state == "open":
            if (
                self.last_failure
                and (time.time() - self.last_failure) >= self.reset_timeout
            ):
                self.state = "half_open"
                return False
            return True
        return False

    @traces(spec="002", story="FR-008", reqs=["REQ-002-FR8-020"])
    def record_success(self) -> None:
        if self.state == "half_open":
            self.state = "closed"
        self.failures = 0

    @traces(spec="002", story="FR-008", reqs=["REQ-002-FR8-020"])
    def record_failure(self) -> None:
        self.failures += 1
        self.last_failure = time.time()
        if self.failures >= self.failure_threshold:
            self.state = "open"


@traces(
    spec="002",
    story="FR-008",
    reqs=[
        "REQ-002-FR8-016",
        "REQ-002-FR8-017",
        "REQ-002-FR8-019",
        "REQ-002-FR8-020",
        "REQ-002-US-001-004",
    ],
)
class PostmanErrorHandler(IErrorHandler):
    """Error handler with circuit breakers and exponential backoff."""

    @traces(spec="002", story="US-001", reqs=["REQ-002-US-001-024"])
    def __init__(self, max_retries: int = 3, failure_threshold: int = 5) -> None:
        self.max_retries = max_retries
        self.failure_threshold = failure_threshold
        self._circuits: dict[str, CircuitBreaker] = {}
        self._lock = asyncio.Lock()

    @traces(
        spec="002",
        story="FR-008",
        reqs=[
            "REQ-002-FR8-016",
            "REQ-002-FR8-017",
            "REQ-002-FR8-019",
            "REQ-002-FR8-020",
        ],
    )
    async def handle(
        self, error: Exception, context: dict[str, Any]
    ) -> tuple[bool, str | None]:
        destination = context.get("destination", "unknown")
        attempt = context.get("retry_attempt", 0)

        # Check circuit breaker
        if self.is_circuit_open(destination):
            return False, "Circuit breaker open"

        # Determine if retry should be attempted
        if attempt >= self.max_retries:
            self.record_failure(destination)
            return False, "Max retries exceeded"

        if isinstance(error, (CircuitOpenError, MaxRetriesExceededError)):
            return False, "Non-retriable error"

        if isinstance(error, DeliveryFailedError):
            return False, "Permanent failure"

        # Calculate exponential backoff with jitter
        delay_ms = self._calculate_backoff(attempt)

        # Record failure for circuit breaker
        self.record_failure(destination)

        return True, f"Retry after {delay_ms:.0f}ms"

    @traces(spec="002", story="FR-008", reqs=["REQ-002-FR8-020"])
    def is_circuit_open(self, destination: str) -> bool:
        circuit = self._get_circuit(destination)
        return circuit.is_open()

    @traces(spec="002", story="FR-008", reqs=["REQ-002-FR8-020"])
    def record_success(self, destination: str) -> None:
        circuit = self._get_circuit(destination)
        circuit.record_success()

    @traces(spec="002", story="FR-008", reqs=["REQ-002-FR8-020"])
    def record_failure(self, destination: str) -> None:
        circuit = self._get_circuit(destination)
        circuit.record_failure()

    @traces(spec="002", story="FR-008", reqs=["REQ-002-FR8-020"])
    def _get_circuit(self, destination: str) -> CircuitBreaker:
        if destination not in self._circuits:
            self._circuits[destination] = CircuitBreaker(
                failure_threshold=self.failure_threshold
            )
        return self._circuits[destination]

    @traces(spec="002", story="US-007", reqs=["REQ-002-US-007-014"])
    def _calculate_backoff(self, attempt: int) -> float:
        base_delay = RETRY_BASE_DELAY_MS * (2**attempt)  # milliseconds
        jitter = random.uniform(-0.1, 0.1) * base_delay
        return max(0, base_delay + jitter)