Write a backoff module handling retries with increasing timeouts

Signed-off-by: camille-bouvy-frequenz <[email protected]>

Author: camille-bouvy-frequenz

src/frequenz/client/electricity_trading/_backoff.py

@@ -0,0 +1,123 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Module to define a backoff strategy to handle retries with increasing timeouts."""
+
+import asyncio
+import logging
+from datetime import datetime, timezone
+from typing import Any, Awaitable, Callable
+
+_logger = logging.getLogger(__name__)
+
+
+class Backoff:
+    """Backoff manager to handle retries with increasing timeouts."""
+
+    def __init__(  # pylint: disable=too-many-arguments, too-many-positional-arguments
+        self,
+        initial_timeout: float,
+        max_timeout: float,
+        timeout_growth_factor: float,
+        max_retries: int,
+        reset_interval: float,
+    ) -> None:
+        """
+        Initialize the backoff manager.
+
+        Args:
+            initial_timeout: The initial timeout duration in seconds.
+            max_timeout: The maximum timeout duration in seconds.
+            timeout_growth_factor: Factor by which to increase the timeout on each retry.
+            max_retries: The maximum number of retry attempts.
+            reset_interval: Time in seconds before resetting the timeout if calls succeed.
+
+        Raises:
+            ValueError: If the timeout growth factor is less than or equal to 1.
+        """
+        if timeout_growth_factor <= 1.0:
+            raise ValueError("timeout_growth_factor must be greater than 1.0")
+        if reset_interval < 0:
+            raise ValueError("reset_interval must be non-negative.")
+        self._initial_timeout = initial_timeout
+        self._timeout = initial_timeout
+        self._max_timeout = max_timeout
+        self._timeout_growth_factor = timeout_growth_factor
+        self._max_retries = max_retries
+        self._reset_interval = reset_interval
+        self._last_success_time = datetime.now(timezone.utc)
+
+    def _increase_timeout(self) -> None:
+        """Increase the timeout using the growth factor, but not beyond the maximum."""
+        if self._timeout < self._max_timeout:
+            self._timeout = min(
+                self._max_timeout, self._timeout * self._timeout_growth_factor
+            )
+            _logger.info("Increased timeout to %ds.", self._timeout)
+        else:
+            _logger.warning(
+                "Timeout is already at the maximum (%ds). No further increments.",
+                self._max_timeout,
+            )
+
+    def _reset_timeout_if_needed(self) -> None:
+        """Reset the timeout to the initial value if the reset interval has passed."""
+        if self._reset_interval == 0:
+            # Skip reset logic if reset_interval is 0
+            return
+
+        now = datetime.now(timezone.utc)
+        if (now - self._last_success_time).total_seconds() >= self._reset_interval:
+            self._timeout = self._initial_timeout
+            _logger.info(
+                "Timeout reset to initial value %ds after %ds seconds without timeout errors.",
+                self._initial_timeout,
+                self._reset_interval,
+            )
+
+    async def execute_with_backoff(
+        self,
+        call: Callable[..., Awaitable[Any]],
+        *args: Any,
+        **kwargs: Any,
+    ) -> Any:
+        """
+        Execute a function with backoff logic for retries on timeout.
+
+        Args:
+            call: The asynchronous function to call.
+            *args: Positional arguments for the function.
+            **kwargs: Keyword arguments for the function.
+
+        Returns:
+            The result of the function call.
+
+        Raises:
+            asyncio.TimeoutError: If all retries are exhausted.
+        """
+        for attempt in range(self._max_retries + 1):
+            try:
+                self._reset_timeout_if_needed()
+                result = await asyncio.wait_for(
+                    call(*args, **kwargs), timeout=self._timeout
+                )
+                # Update last success time on successful call
+                self._last_success_time = datetime.now(timezone.utc)
+                return result
+            except asyncio.TimeoutError:
+                call_name = call.__name__ if hasattr(call, "__name__") else str(call)
+                if attempt == self._max_retries:
+                    _logger.error(
+                        "Timeout after %d retries (timeout=%ds): %s",
+                        self._max_retries,
+                        self._timeout,
+                        call,
+                    )
+                    raise
+                _logger.warning(
+                    "Timeout on %s attempt %d/%d...",
+                    call_name,
+                    attempt + 1,
+                    self._max_retries + 1,
+                )
+                self._increase_timeout()

src/frequenz/client/electricity_trading/_client.py

@@ -3,6 +3,8 @@
 
 """Module to define the client class."""
 
+# pylint: disable=too-many-lines
+
 from __future__ import annotations
 
 import logging
@@ -27,6 +29,7 @@
 from frequenz.client.common.pagination import Params
 from google.protobuf import field_mask_pb2, struct_pb2
 
+from ._backoff import Backoff
 from ._types import (
     DeliveryArea,
     DeliveryPeriod,
@@ -91,7 +94,9 @@ def validate_decimal_places(value: Decimal, decimal_places: int, name: str) -> N
         ) from exc
 
 
-class Client(BaseApiClient[ElectricityTradingServiceStub]):
+class Client(
+    BaseApiClient[ElectricityTradingServiceStub]
+):  # pylint:disable=too-many-instance-attributes
     """Electricity trading client."""
 
     _instances: dict[tuple[str, str | None], "Client"] = {}
@@ -120,15 +125,28 @@ def __new__(
 
         return cls._instances[key]
 
-    def __init__(
-        self, server_url: str, connect: bool = True, auth_key: str | None = None
+    def __init__(  # pylint: disable=too-many-arguments, too-many-positional-arguments
+        self,
+        server_url: str,
+        connect: bool = True,
+        auth_key: str | None = None,
+        initial_timeout: float = 10.0,
+        max_timeout: float = 300.0,
+        timeout_growth_factor: float = 2.0,
+        max_timeout_retries: int = 5,
+        reset_interval: int = 600,
     ) -> None:
         """Initialize the client.
 
         Args:
             server_url: The URL of the Electricity Trading service.
             connect: Whether to connect to the server immediately.
             auth_key: The API key for the authorization.
+            initial_timeout: Initial timeout duration for gRPC calls (in seconds).
+            max_timeout: Maximum timeout duration for gRPC calls (in seconds).
+            timeout_growth_factor: Factor by which to increase the timeout on each retry.
+            max_timeout_retries: Maximum number of retry attempts when a timeout is reached.
+            reset_interval: Time (in seconds) before resetting timeout if everything behaves fine.
         """
         if not hasattr(
             self, "_initialized"
@@ -163,6 +181,15 @@ def __init__(
 
         self._metadata = (("key", auth_key),) if auth_key else ()
 
+        # Backoff strategy configuration to handle gRPC connection failures
+        self._backoff = Backoff(
+            initial_timeout=initial_timeout,
+            max_timeout=max_timeout,
+            timeout_growth_factor=timeout_growth_factor,
+            max_retries=max_timeout_retries,
+            reset_interval=reset_interval,
+        )
+
     @property
     def stub(self) -> electricity_trading_pb2_grpc.ElectricityTradingServiceAsyncStub:
         """
@@ -503,10 +530,10 @@ async def create_gridpool_order(
         try:
             response = await cast(
                 Awaitable[electricity_trading_pb2.CreateGridpoolOrderResponse],
-                self.stub.CreateGridpoolOrder(
+                self._backoff.execute_with_backoff(
+                    self.stub.CreateGridpoolOrder,
                     electricity_trading_pb2.CreateGridpoolOrderRequest(
-                        gridpool_id=gridpool_id,
-                        order=order.to_pb(),
+                        gridpool_id=gridpool_id, order=order.to_pb()
                     ),
                     metadata=self._metadata,
                 ),
@@ -615,7 +642,8 @@ async def update_gridpool_order(
         try:
             response = await cast(
                 Awaitable[electricity_trading_pb2.UpdateGridpoolOrderResponse],
-                self.stub.UpdateGridpoolOrder(
+                self._backoff.execute_with_backoff(
+                    self.stub.UpdateGridpoolOrder,
                     electricity_trading_pb2.UpdateGridpoolOrderRequest(
                         gridpool_id=gridpool_id,
                         order_id=order_id,
@@ -650,7 +678,8 @@ async def cancel_gridpool_order(
         try:
             response = await cast(
                 Awaitable[electricity_trading_pb2.CancelGridpoolOrderResponse],
-                self.stub.CancelGridpoolOrder(
+                self._backoff.execute_with_backoff(
+                    self.stub.CancelGridpoolOrder,
                     electricity_trading_pb2.CancelGridpoolOrderRequest(
                         gridpool_id=gridpool_id, order_id=order_id
                     ),
@@ -678,7 +707,8 @@ async def cancel_all_gridpool_orders(self, gridpool_id: int) -> int:
         try:
             response = await cast(
                 Awaitable[electricity_trading_pb2.CancelAllGridpoolOrdersResponse],
-                self.stub.CancelAllGridpoolOrders(
+                self._backoff.execute_with_backoff(
+                    self.stub.CancelAllGridpoolOrders,
                     electricity_trading_pb2.CancelAllGridpoolOrdersRequest(
                         gridpool_id=gridpool_id
                     ),
@@ -710,7 +740,8 @@ async def get_gridpool_order(self, gridpool_id: int, order_id: int) -> OrderDeta
         try:
             response = await cast(
                 Awaitable[electricity_trading_pb2.GetGridpoolOrderResponse],
-                self.stub.GetGridpoolOrder(
+                self._backoff.execute_with_backoff(
+                    self.stub.GetGridpoolOrder,
                     electricity_trading_pb2.GetGridpoolOrderRequest(
                         gridpool_id=gridpool_id, order_id=order_id
                     ),
@@ -770,7 +801,8 @@ async def list_gridpool_orders(
         try:
             response = await cast(
                 Awaitable[electricity_trading_pb2.ListGridpoolOrdersResponse],
-                self.stub.ListGridpoolOrders(
+                self._backoff.execute_with_backoff(
+                    self.stub.ListGridpoolOrders,
                     electricity_trading_pb2.ListGridpoolOrdersRequest(
                         gridpool_id=gridpool_id,
                         filter=gridpool_order_filer.to_pb(),
@@ -842,7 +874,8 @@ async def list_gridpool_trades(
         try:
             response = await cast(
                 Awaitable[electricity_trading_pb2.ListGridpoolTradesResponse],
-                self.stub.ListGridpoolTrades(
+                self._backoff.execute_with_backoff(
+                    self.stub.ListGridpoolTrades,
                     electricity_trading_pb2.ListGridpoolTradesRequest(
                         gridpool_id=gridpool_id,
                         filter=gridpool_trade_filter.to_pb(),
@@ -899,7 +932,8 @@ async def list_public_trades(
         try:
             response = await cast(
                 Awaitable[electricity_trading_pb2.ListPublicTradesResponse],
-                self.stub.ListPublicTrades(
+                self._backoff.execute_with_backoff(
+                    self.stub.ListPublicTrades,
                     electricity_trading_pb2.ListPublicTradesRequest(
                         filter=public_trade_filter.to_pb(),
                         pagination_params=pagination_params.to_proto(),