"""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()