diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d4de0b74..6be017e5a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +### Added + +- Added `exponential_backoff_time()` to `utils.network` module. + ### Fixed - Fixed an issue that would cause `DeepgramSTTService` to stop working after an diff --git a/src/pipecat/services/websocket_service.py b/src/pipecat/services/websocket_service.py index c47d6d97e..a8dc97414 100644 --- a/src/pipecat/services/websocket_service.py +++ b/src/pipecat/services/websocket_service.py @@ -13,6 +13,7 @@ from loguru import logger from websockets.protocol import State from pipecat.frames.frames import ErrorFrame +from pipecat.utils.network import exponential_backoff_time class WebsocketService(ABC): @@ -51,27 +52,6 @@ class WebsocketService(ABC): await self._connect_websocket() return await self._verify_connection() - def _calculate_wait_time( - self, attempt: int, min_wait: float = 4, max_wait: float = 10, multiplier: float = 1 - ) -> float: - """Calculate exponential backoff wait time. - - Args: - attempt: Current attempt number (1-based) - min_wait: Minimum wait time in seconds - max_wait: Maximum wait time in seconds - multiplier: Base multiplier for exponential calculation - - Returns: - Wait time in seconds - """ - try: - exp = 2 ** (attempt - 1) * multiplier - result = max(0, min(exp, max_wait)) - return max(min_wait, result) - except (ValueError, ArithmeticError): - return max_wait - async def _receive_task_handler(self, report_error: Callable[[ErrorFrame], Awaitable[None]]): """Handles WebSocket message receiving with automatic retry logic. @@ -104,7 +84,7 @@ class WebsocketService(ABC): try: if await self._reconnect_websocket(retry_count): retry_count = 0 # Reset counter on successful reconnection - wait_time = self._calculate_wait_time(retry_count) + wait_time = exponential_backoff_time(retry_count) await asyncio.sleep(wait_time) except Exception as reconnect_error: logger.error(f"{self} reconnection failed: {reconnect_error}") diff --git a/src/pipecat/utils/network.py b/src/pipecat/utils/network.py new file mode 100644 index 000000000..27bec990c --- /dev/null +++ b/src/pipecat/utils/network.py @@ -0,0 +1,27 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +def exponential_backoff_time( + attempt: int, min_wait: float = 4, max_wait: float = 10, multiplier: float = 1 +) -> float: + """Calculate exponential backoff wait time. + + Args: + attempt: Current attempt number (1-based) + min_wait: Minimum wait time in seconds + max_wait: Maximum wait time in seconds + multiplier: Base multiplier for exponential calculation + + Returns: + Wait time in seconds + """ + try: + exp = 2 ** (attempt - 1) * multiplier + result = max(0, min(exp, max_wait)) + return max(min_wait, result) + except (ValueError, ArithmeticError): + return max_wait diff --git a/tests/test_utils_network.py b/tests/test_utils_network.py new file mode 100644 index 000000000..5b8d96939 --- /dev/null +++ b/tests/test_utils_network.py @@ -0,0 +1,34 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import unittest + +from pipecat.utils.network import exponential_backoff_time + + +class TestUtilsNetwork(unittest.IsolatedAsyncioTestCase): + async def test_exponential_backoff_time(self): + # min_wait=4, max_wait=10, multiplier=1 + assert exponential_backoff_time(attempt=1, min_wait=4, max_wait=10, multiplier=1) == 4 + assert exponential_backoff_time(attempt=2, min_wait=4, max_wait=10, multiplier=1) == 4 + assert exponential_backoff_time(attempt=3, min_wait=4, max_wait=10, multiplier=1) == 4 + assert exponential_backoff_time(attempt=4, min_wait=4, max_wait=10, multiplier=1) == 8 + assert exponential_backoff_time(attempt=5, min_wait=4, max_wait=10, multiplier=1) == 10 + assert exponential_backoff_time(attempt=6, min_wait=4, max_wait=10, multiplier=1) == 10 + # min_wait=1, max_wait=10, multiplier=1 + assert exponential_backoff_time(attempt=1, min_wait=1, max_wait=10, multiplier=1) == 1 + assert exponential_backoff_time(attempt=2, min_wait=1, max_wait=10, multiplier=1) == 2 + assert exponential_backoff_time(attempt=3, min_wait=1, max_wait=10, multiplier=1) == 4 + assert exponential_backoff_time(attempt=4, min_wait=1, max_wait=10, multiplier=1) == 8 + assert exponential_backoff_time(attempt=5, min_wait=1, max_wait=10, multiplier=1) == 10 + assert exponential_backoff_time(attempt=6, min_wait=1, max_wait=10, multiplier=1) == 10 + # min_wait=1, max_wait=20, multiplier=2 + assert exponential_backoff_time(attempt=1, min_wait=1, max_wait=20, multiplier=2) == 2 + assert exponential_backoff_time(attempt=2, min_wait=1, max_wait=20, multiplier=2) == 4 + assert exponential_backoff_time(attempt=3, min_wait=1, max_wait=20, multiplier=2) == 8 + assert exponential_backoff_time(attempt=4, min_wait=1, max_wait=20, multiplier=2) == 16 + assert exponential_backoff_time(attempt=5, min_wait=1, max_wait=20, multiplier=2) == 20 + assert exponential_backoff_time(attempt=6, min_wait=1, max_wait=20, multiplier=2) == 20