cleveragents-core/benchmarks/core_circuit_breaker_bench.py

"""ASV benchmarks for circuit breaker pattern.

Measures the overhead of circuit breaker state management,
state transitions, and fast-fail latency in open state.
"""

from __future__ import annotations

import asyncio
import time

from cleveragents.core.circuit_breaker import (
    CircuitBreaker,
    CircuitBreakerOpen,
    CircuitBreakerState,
)


class CircuitBreakerClosedStateBench:
    """Benchmark circuit breaker overhead in closed state."""

    timeout = 60

    def setup(self) -> None:
        """Set up circuit breaker in closed state."""
        self.breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60.0,
            expected_exception=Exception,
            half_open_max_successes=2,
            cooldown_seconds=30.0,
            name="test_service",
        )

    def time_call_success(self) -> None:
        """Benchmark successful call overhead in closed state."""

        def dummy_func() -> str:
            return "success"

        self.breaker.call(dummy_func)

    def time_call_with_args(self) -> None:
        """Benchmark call with arguments in closed state."""

        def dummy_func(x: int, y: str) -> str:
            return f"{x}:{y}"

        self.breaker.call(dummy_func, 42, "test")

    def time_state_check(self) -> None:
        """Benchmark state property access."""
        _ = self.breaker.state

    def time_failure_count_check(self) -> None:
        """Benchmark failure count property access."""
        _ = self.breaker.failure_count


class CircuitBreakerOpenStateBench:
    """Benchmark circuit breaker fast-fail latency in open state."""

    timeout = 60

    def setup(self) -> None:
        """Set up circuit breaker in open state."""
        self.breaker = CircuitBreaker(
            failure_threshold=2,
            recovery_timeout=60.0,
            expected_exception=Exception,
            half_open_max_successes=2,
            cooldown_seconds=30.0,
            name="test_service",
        )
        # Force the breaker into open state
        for _ in range(3):
            try:
                self.breaker.call(lambda: 1 / 0)  # Raise ZeroDivisionError
            except ZeroDivisionError:
                pass

    def time_fast_fail(self) -> None:
        """Benchmark fast-fail latency when circuit is open."""
        try:
            self.breaker.call(lambda: "should not execute")
        except CircuitBreakerOpen:
            pass

    def time_open_state_check(self) -> None:
        """Benchmark state check when open."""
        assert self.breaker.state == CircuitBreakerState.OPEN


class CircuitBreakerClosedToOpenTransitionBench:
    """Benchmark closed-to-open state transition overhead."""

    timeout = 60

    def setup(self) -> None:
        """Set up a fresh circuit breaker in closed state."""
        self.breaker = CircuitBreaker(
            failure_threshold=2,
            recovery_timeout=60.0,
            expected_exception=Exception,
            half_open_max_successes=1,
            cooldown_seconds=30.0,
            name="test_service",
        )

    def time_closed_to_open_transition(self) -> None:
        """Benchmark transition from closed to open state."""
        # Trigger failures to transition to open
        for _ in range(3):
            try:
                self.breaker.call(lambda: 1 / 0)
            except ZeroDivisionError:
                pass


class CircuitBreakerOpenToHalfOpenTransitionBench:
    """Benchmark open-to-half-open state transition overhead."""

    timeout = 60

    def setup(self) -> None:
        """Set up circuit breaker already in open state."""
        self.breaker = CircuitBreaker(
            failure_threshold=2,
            recovery_timeout=0.05,  # Very short timeout
            expected_exception=Exception,
            half_open_max_successes=1,
            cooldown_seconds=0.01,
            name="test_service",
        )
        # Force to open state
        for _ in range(3):
            try:
                self.breaker.call(lambda: 1 / 0)
            except ZeroDivisionError:
                pass
        # Wait for recovery timeout so transition to half-open is possible
        time.sleep(0.1)

    def time_open_to_half_open_transition(self) -> None:
        """Benchmark transition from open to half-open state."""
        # Attempt call to trigger half-open transition
        try:
            self.breaker.call(lambda: "success")
        except CircuitBreakerOpen:
            pass


class CircuitBreakerHalfOpenToClosedTransitionBench:
    """Benchmark half-open-to-closed state transition overhead."""

    timeout = 60

    def setup(self) -> None:
        """Set up circuit breaker already in half-open state."""
        self.breaker = CircuitBreaker(
            failure_threshold=2,
            recovery_timeout=0.05,
            expected_exception=Exception,
            half_open_max_successes=1,
            cooldown_seconds=0.01,
            name="test_service",
        )
        # Force to open state
        for _ in range(3):
            try:
                self.breaker.call(lambda: 1 / 0)
            except ZeroDivisionError:
                pass
        # Wait for recovery timeout so breaker enters half-open on next call
        time.sleep(0.1)

    def time_half_open_to_closed_transition(self) -> None:
        """Benchmark transition from half-open to closed state."""
        # Successful call in half-open should close the breaker
        self.breaker.call(lambda: "success")


class CircuitBreakerAsyncBench:
    """Benchmark async circuit breaker operations."""

    timeout = 60

    def setup(self) -> None:
        """Set up async circuit breakers."""
        self.breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60.0,
            expected_exception=Exception,
            half_open_max_successes=2,
            cooldown_seconds=30.0,
            name="test_service",
        )
        # Set up a separate breaker already in open state for fast-fail benchmark
        self.open_breaker = CircuitBreaker(
            failure_threshold=2,
            recovery_timeout=60.0,
            expected_exception=Exception,
            half_open_max_successes=2,
            cooldown_seconds=30.0,
            name="test_service_open",
        )
        for _ in range(3):
            try:
                self.open_breaker.call(lambda: 1 / 0)
            except ZeroDivisionError:
                pass

    def time_async_call_success(self) -> None:
        """Benchmark successful async call overhead."""

        async def dummy_async() -> str:
            return "success"

        asyncio.run(self.breaker.async_call(dummy_async))

    def time_async_call_with_args(self) -> None:
        """Benchmark async call with arguments."""

        async def dummy_async(x: int, y: str) -> str:
            return f"{x}:{y}"

        asyncio.run(self.breaker.async_call(dummy_async, 42, "test"))

    def time_async_fast_fail(self) -> None:
        """Benchmark async fast-fail when open."""

        async def dummy_async() -> str:
            return "should not execute"

        try:
            asyncio.run(self.open_breaker.async_call(dummy_async))
        except CircuitBreakerOpen:
            pass


class CircuitBreakerInitializationBench:
    """Benchmark circuit breaker initialization overhead."""

    timeout = 60

    def time_default_initialization(self) -> None:
        """Benchmark default initialization."""
        CircuitBreaker()

    def time_custom_initialization(self) -> None:
        """Benchmark initialization with custom parameters."""
        CircuitBreaker(
            failure_threshold=10,
            recovery_timeout=120.0,
            expected_exception=ValueError,
            half_open_max_successes=3,
            cooldown_seconds=45.0,
            name="custom_service",
        )

    def time_multiple_breakers(self) -> None:
        """Benchmark creating multiple breakers."""
        for i in range(10):
            CircuitBreaker(name=f"service_{i}")