cleveragents-core/benchmarks/core_retry_patterns_bench.py

"""ASV benchmarks for retry patterns.

Measures the overhead of retry decorator construction and
happy-path invocation for the public retry factory functions.
"""

from __future__ import annotations

import asyncio

from cleveragents.core.retry_patterns import (
    get_retry_decorator,
    retry_on_result,
    retry_with_exponential_backoff,
    retry_with_jitter,
    retry_with_timeout,
    should_retry_result,
)


class RetryDecoratorConstructionBench:
    """Benchmark retry decorator construction overhead."""

    timeout = 60

    def time_retry_with_exponential_backoff_construction(self) -> None:
        """Benchmark constructing exponential backoff decorator."""

        @retry_with_exponential_backoff(max_attempts=3)
        def dummy_func() -> str:
            return "success"

    def time_retry_with_jitter_construction(self) -> None:
        """Benchmark constructing jitter decorator."""

        @retry_with_jitter(max_attempts=3)
        def dummy_func() -> str:
            return "success"

    def time_retry_with_timeout_construction(self) -> None:
        """Benchmark constructing timeout decorator."""

        @retry_with_timeout(timeout_seconds=5.0)
        def dummy_func() -> str:
            return "success"

    def time_retry_on_result_construction(self) -> None:
        """Benchmark constructing result-based retry decorator."""

        @retry_on_result(should_retry_result, max_attempts=3)
        def dummy_func() -> str:
            return "success"

    def time_get_retry_decorator_network(self) -> None:
        """Benchmark constructing decorator via factory for network category."""
        get_retry_decorator("network")

    def time_get_retry_decorator_provider(self) -> None:
        """Benchmark constructing decorator via factory for provider category."""
        get_retry_decorator("provider")

    def time_get_retry_decorator_database(self) -> None:
        """Benchmark constructing decorator via factory for database category."""
        get_retry_decorator("database")

    def time_get_retry_decorator_unknown(self) -> None:
        """Benchmark constructing decorator via factory for unknown category."""
        get_retry_decorator("unknown_category")


class RetryDecoratorInvocationBench:
    """Benchmark happy-path retry decorator invocation."""

    timeout = 60

    def setup(self) -> None:
        """Set up decorated functions."""

        @retry_with_exponential_backoff(max_attempts=3)
        def exponential_func() -> str:
            return "success"

        @retry_with_jitter(max_attempts=3)
        def jitter_func() -> str:
            return "success"

        @retry_with_timeout(timeout_seconds=5.0)
        def timeout_func() -> str:
            return "success"

        @retry_on_result(should_retry_result, max_attempts=3)
        def result_func() -> str:
            return "success"

        self.exponential_func = exponential_func
        self.jitter_func = jitter_func
        self.timeout_func = timeout_func
        self.result_func = result_func

    def time_exponential_backoff_invocation(self) -> None:
        """Benchmark exponential backoff invocation (happy path)."""
        self.exponential_func()

    def time_jitter_invocation(self) -> None:
        """Benchmark jitter invocation (happy path)."""
        self.jitter_func()

    def time_timeout_invocation(self) -> None:
        """Benchmark timeout invocation (happy path)."""
        self.timeout_func()

    def time_result_based_invocation(self) -> None:
        """Benchmark result-based retry invocation (happy path)."""
        self.result_func()


class RetryDecoratorAsyncBench:
    """Benchmark async retry decorator operations."""

    timeout = 60

    def setup(self) -> None:
        """Set up async decorated functions."""

        @retry_with_exponential_backoff(max_attempts=3)
        async def async_exponential() -> str:
            return "success"

        @retry_with_jitter(max_attempts=3)
        async def async_jitter() -> str:
            return "success"

        @retry_with_timeout(timeout_seconds=5.0)
        async def async_timeout() -> str:
            return "success"

        self.async_exponential = async_exponential
        self.async_jitter = async_jitter
        self.async_timeout = async_timeout

    def time_async_exponential_invocation(self) -> None:
        """Benchmark async exponential backoff invocation."""
        asyncio.run(self.async_exponential())

    def time_async_jitter_invocation(self) -> None:
        """Benchmark async jitter invocation."""
        asyncio.run(self.async_jitter())

    def time_async_timeout_invocation(self) -> None:
        """Benchmark async timeout invocation."""
        asyncio.run(self.async_timeout())


class RetryDecoratorWithArgumentsBench:
    """Benchmark retry decorators with function arguments."""

    timeout = 60

    def setup(self) -> None:
        """Set up decorated functions with arguments."""

        @retry_with_exponential_backoff(max_attempts=3)
        def func_with_args(x: int, y: str, z: float = 1.0) -> str:
            return f"{x}:{y}:{z}"

        @retry_with_jitter(max_attempts=3)
        def func_with_kwargs(a: int, b: str = "default") -> str:
            return f"{a}:{b}"

        self.func_with_args = func_with_args
        self.func_with_kwargs = func_with_kwargs

    def time_exponential_with_positional_args(self) -> None:
        """Benchmark exponential backoff with positional arguments."""
        self.func_with_args(42, "test", 2.5)

    def time_exponential_with_keyword_args(self) -> None:
        """Benchmark exponential backoff with keyword arguments."""
        self.func_with_args(x=42, y="test", z=2.5)

    def time_jitter_with_mixed_args(self) -> None:
        """Benchmark jitter with mixed positional and keyword arguments."""
        self.func_with_kwargs(100, b="custom")


class RetryDecoratorFactoryBench:
    """Benchmark retry decorator factory function."""

    timeout = 60

    def time_factory_network_category(self) -> None:
        """Benchmark factory with network category."""
        decorator = get_retry_decorator("network")

        @decorator
        def dummy() -> str:
            return "success"

        dummy()

    def time_factory_provider_category(self) -> None:
        """Benchmark factory with provider category."""
        decorator = get_retry_decorator("provider")

        @decorator
        def dummy() -> str:
            return "success"

        dummy()

    def time_factory_database_category(self) -> None:
        """Benchmark factory with database category."""
        decorator = get_retry_decorator("database")

        @decorator
        def dummy() -> str:
            return "success"

        dummy()

    def time_factory_file_operation_category(self) -> None:
        """Benchmark factory with file_operation category."""
        decorator = get_retry_decorator("file_operation")

        @decorator
        def dummy() -> str:
            return "success"

        dummy()


class RetryDecoratorConfigurationBench:
    """Benchmark retry decorator with various configurations."""

    timeout = 60

    def time_minimal_config(self) -> None:
        """Benchmark decorator with minimal configuration."""

        @retry_with_exponential_backoff()
        def dummy() -> str:
            return "success"

        dummy()

    def time_aggressive_config(self) -> None:
        """Benchmark decorator with aggressive retry settings."""

        @retry_with_exponential_backoff(max_attempts=10)
        def dummy() -> str:
            return "success"

        dummy()

    def time_conservative_config(self) -> None:
        """Benchmark decorator with conservative retry settings."""

        @retry_with_exponential_backoff(max_attempts=1)
        def dummy() -> str:
            return "success"

        dummy()

    def time_long_timeout_config(self) -> None:
        """Benchmark decorator with long timeout."""

        @retry_with_timeout(timeout_seconds=300.0)
        def dummy() -> str:
            return "success"

        dummy()

    def time_short_timeout_config(self) -> None:
        """Benchmark decorator with short timeout."""

        @retry_with_timeout(timeout_seconds=0.1)
        def dummy() -> str:
            return "success"

        dummy()