orchestrata_ultimate.providers.circuit_breaker

Reference › orchestrata_ultimate › providers › circuit_breaker

circuit_breaker isolates failing providers (specification 011, user story US-003). CircuitBreaker is created per provider by provider_manager and read by the router. It has three states in CircuitState: closed, open and half-open. In the closed state record_failure counts failures and opens the circuit at failure_threshold (default 5). After cooldown_seconds (default 60) the open circuit moves to half-open (_maybe_cooldown_transition, applied whenever state or is_available is read). In half-open a failure reopens the circuit at once, and success_threshold successes (default 3) close it again. State changes are protected by a lock. With a Redis client from set_redis, _save_state persists the state and _load_state restores it at startup, so breaker state survives restarts. reset forces the circuit closed.

At a glance

Class diagram

Dependency diagram

API

Classes

CircuitState (enum)

Bases: StrEnum · Decorators: traces(spec='011', story='US-003', reqs=['REQ-011-038']) · Line 37

CircuitBreaker

Decorators: traces(spec='011', story='US-003', reqs=['REQ-011-038', 'REQ-011-039', 'REQ-011-040']) · Line 47

Per-provider circuit breaker with Redis persistence.

Functions

Source

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

# mypy: ignore-errors
"""Per-provider circuit breaker.

SPEC-011 FR-009: Implements closed -> open -> half-open state machine
with configurable failure threshold, cooldown, and success threshold.
State persisted in Redis for restart recovery.
"""

import logging
import threading
import time
from collections.abc import Awaitable
from enum import StrEnum
from typing import cast

import redis.asyncio as redis

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)


def _redis_result[T](value: Awaitable[T] | T) -> Awaitable[T]:
    """Narrow a redis-py command result back to its true async type.

    redis-py types command methods (``hset``, ``hgetall``, ...) as
    ``Awaitable[T] | T`` because the same command mixin backs both the sync
    and async clients (see ``redis.typing.ResponseT``). ``CircuitBreaker``
    only ever holds an async client, which always returns an awaitable at
    runtime -- this documents that fact for pyright instead of suppressing
    the check.
    """
    return cast(Awaitable[T], value)


@traces(spec="011", story="US-003", reqs=["REQ-011-038"])
class CircuitState(StrEnum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"


_CB_PREFIX = "orchestrata:provider:cb"


@traces(spec="011", story="US-003", reqs=["REQ-011-038", "REQ-011-039", "REQ-011-040"])
class CircuitBreaker:
    """Per-provider circuit breaker with Redis persistence.

    State machine: CLOSED --(failures >= threshold)--> OPEN
                   OPEN --(cooldown elapsed)--> HALF_OPEN
                   HALF_OPEN --(success >= success_threshold)--> CLOSED
                   HALF_OPEN --(failure)--> OPEN
    """

    @traces(spec="011", story="US-003", reqs=["REQ-011-039", "REQ-011-040"])
    def __init__(
        self,
        provider_name: str,
        failure_threshold: int = 5,
        cooldown_seconds: float = 60.0,
        success_threshold: int = 3,
    ):
        self.provider_name = provider_name
        self.failure_threshold = failure_threshold
        self.cooldown_seconds = cooldown_seconds
        self.success_threshold = success_threshold
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._success_count = 0
        self._last_failure_time: float | None = None
        self._redis: redis.Redis | None = None
        # Guards all reads+mutations of the counters and state so concurrent
        # callers (free-threaded 3.14, no GIL) cannot interleave a check and a
        # mutation and corrupt the failure/success counts or the state machine.
        self._lock = threading.Lock()

    def _maybe_cooldown_transition(self) -> None:
        """Apply OPEN -> HALF_OPEN if cooldown elapsed. Caller holds the lock."""
        if self._state == CircuitState.OPEN and self._last_failure_time:
            elapsed = time.monotonic() - self._last_failure_time
            if elapsed >= self.cooldown_seconds:
                self._state = CircuitState.HALF_OPEN
                self._success_count = 0

    @property
    @traces(spec="011", story="US-003", reqs=["REQ-011-038", "REQ-011-040"])
    def state(self) -> CircuitState:
        """Current circuit state, accounting for cooldown transitions."""
        with self._lock:
            self._maybe_cooldown_transition()
            return self._state

    @property
    @traces(spec="011", story="US-003", reqs=["REQ-011-038", "REQ-011-040"])
    def is_available(self) -> bool:
        """Whether the circuit allows requests."""
        with self._lock:
            self._maybe_cooldown_transition()
            return self._state != CircuitState.OPEN

    @traces(spec="011", story="US-003", reqs=["REQ-011-038"])
    async def set_redis(self, client: redis.Redis) -> None:
        """Set Redis client for state persistence."""
        self._redis = client
        await self._load_state()

    @traces(spec="011", story="US-003", reqs=["REQ-011-038"])
    async def record_success(self) -> None:
        """Record a successful request."""
        with self._lock:
            self._maybe_cooldown_transition()
            if self._state == CircuitState.HALF_OPEN:
                self._success_count += 1
                if self._success_count >= self.success_threshold:
                    self._state = CircuitState.CLOSED
                    self._failure_count = 0
                    self._success_count = 0
                    self._last_failure_time = None
                    logger.info("Circuit %s: HALF_OPEN -> CLOSED", self.provider_name)
            elif self._state == CircuitState.CLOSED:
                self._failure_count = 0
        await self._save_state()

    @traces(spec="011", story="US-003", reqs=["REQ-011-039"])
    async def record_failure(self) -> None:
        """Record a failed request."""
        with self._lock:
            self._maybe_cooldown_transition()
            if self._state == CircuitState.HALF_OPEN:
                self._state = CircuitState.OPEN
                self._last_failure_time = time.monotonic()
                self._success_count = 0
                logger.warning("Circuit %s: HALF_OPEN -> OPEN", self.provider_name)
            elif self._state == CircuitState.CLOSED:
                self._failure_count += 1
                if self._failure_count >= self.failure_threshold:
                    self._state = CircuitState.OPEN
                    self._last_failure_time = time.monotonic()
                    logger.warning(
                        "Circuit %s: CLOSED -> OPEN (failures=%d)",
                        self.provider_name,
                        self._failure_count,
                    )
        await self._save_state()

    async def _save_state(self) -> None:
        """Persist state to Redis."""
        if not self._redis:
            return
        key = f"{_CB_PREFIX}:{self.provider_name}"
        with self._lock:
            mapping = {
                "state": self._state.value,
                "failures": str(self._failure_count),
                "successes": str(self._success_count),
                "last_fail": str(self._last_failure_time or ""),
            }
        await _redis_result(self._redis.hset(key, mapping=mapping))

    @traces(spec="011", story="US-003", reqs=["REQ-011-038"])
    async def _load_state(self) -> None:
        """Load state from Redis on startup."""
        if not self._redis:
            return
        key = f"{_CB_PREFIX}:{self.provider_name}"
        data = await _redis_result(self._redis.hgetall(key))
        if not data:
            return
        state_val = data.get("state", "closed")
        if isinstance(state_val, bytes):
            state_val = state_val.decode()
        self._state = CircuitState(state_val)
        failures = data.get("failures", "0")
        if isinstance(failures, bytes):
            failures = failures.decode()
        self._failure_count = int(failures)
        successes = data.get("successes", "0")
        if isinstance(successes, bytes):
            successes = successes.decode()
        self._success_count = int(successes)
        # On restart, transition OPEN -> HALF_OPEN to allow probe
        if self._state == CircuitState.OPEN:
            self._state = CircuitState.HALF_OPEN
            self._success_count = 0
            logger.info(
                "Circuit %s: OPEN -> HALF_OPEN (restart recovery)",
                self.provider_name,
            )
            await self._save_state()

    def reset(self) -> None:
        """Reset circuit to closed state."""
        with self._lock:
            self._state = CircuitState.CLOSED
            self._failure_count = 0
            self._success_count = 0
            self._last_failure_time = None