Perform jittered retries for ESP-ME* devices on 503

pyrainbird/async_client.py

@@ -15,10 +15,12 @@
 import logging
 from collections.abc import Callable
 from http import HTTPStatus
-from typing import Any, Optional, TypeVar
+from typing import Any, TypeVar, Union
 
 import aiohttp
 from aiohttp.client_exceptions import ClientError, ClientResponseError
+from aiohttp_retry import RetryClient, RetryOptions, JitterRetry
+
 
 from . import encryption, rainbird
 from .data import (
@@ -66,6 +68,16 @@
 DATA = "data"
 CLOUD_API_URL = "http://rdz-rbcloud.rainbird.com/phone-api"
 
+# In general, these devices can handle only one in flight request at a time
+# otherwise return a 503. The caller is expected to follow that, however ESP
+# ME devices also seem to return 503s more regularly than other devices so we
+# include retry behavior for them. We only retry the specific device busy error.
+DEVICE_BUSY_RETRY = JitterRetry(
+    attempts=3,
+    statuses=[HTTPStatus.SERVICE_UNAVAILABLE.value],
+    retry_all_server_errors=False,
+)
+
 
 class AsyncRainbirdClient:
     """An asyncio rainbird client.
@@ -77,17 +89,27 @@ def __init__(
         self,
         websession: aiohttp.ClientSession,
         host: str,
-        password: Optional[str],
+        password: Union[str, None],
     ) -> None:
         self._websession = websession
+        self._host = host
         if host.startswith("/") or host.startswith("http://"):
             self._url = host
         else:
             self._url = f"http://{host}/stick"
+        self._password = password
         self._coder = encryption.PayloadCoder(password, _LOGGER)
 
+    def with_retry_options(self, retry_options: RetryOptions) -> "AsyncRainbirdClient":
+        """Create a new AsyncRainbirdClient with retry options."""
+        return AsyncRainbirdClient(
+            RetryClient(client_session=self._websession, retry_options=retry_options),
+            self._host,
+            self._password,
+        )
+
     async def request(
-        self, method: str, params: dict[str, Any] = None
+        self, method: str, params: Union[dict[str, Any], None] = None
     ) -> dict[str, Any]:
         """Send a request for any command."""
         payload = self._coder.encode_command(method, params or {})
@@ -118,7 +140,7 @@ def CreateController(
     websession: aiohttp.ClientSession, host: str, password: str
 ) -> "AsyncRainbirdController":
     """Create an AsyncRainbirdController."""
-    local_client = AsyncRainbirdClient(websession, host, password)
+    local_client = (AsyncRainbirdClient(websession, host, password),)
     cloud_client = AsyncRainbirdClient(websession, CLOUD_API_URL, None)
     return AsyncRainbirdController(local_client, cloud_client)
 
@@ -135,17 +157,25 @@ def __init__(
         self._local_client = local_client
         self._cloud_client = cloud_client
         self._cache: dict[str, Any] = {}
+        self._model: ModelAndVersion | None = None
 
     async def get_model_and_version(self) -> ModelAndVersion:
         """Return the model and version."""
-        return await self._cacheable_command(
+        response = await self._cacheable_command(
             lambda response: ModelAndVersion(
                 response["modelID"],
                 response["protocolRevisionMajor"],
                 response["protocolRevisionMinor"],
             ),
             "ModelAndVersionRequest",
         )
+        if self._model is None:
+            self._model = response
+            if self._model.model_info.retries:
+                self._local_client = self._local_client.with_retry_options(
+                    DEVICE_BUSY_RETRY
+                )
+        return response
 
     async def get_available_stations(self) -> AvailableStations:
         """Get the available stations."""

pyrainbird/data.py

@@ -38,7 +38,6 @@ class CommandSupport:
     support: int
     """Return if the command is supported."""
 
-
     echo: int
     """Return the input command."""
 
@@ -68,6 +67,9 @@ class ModelInfo:
     max_run_times: int
     """The maximum number of run times supported by the device."""
 
+    retries: bool = False
+    """If device busy errors should be retried"""
+
 
 @dataclass
 class ModelAndVersion:

pyrainbird/resources/models.yaml

@@ -12,6 +12,7 @@
   supports_water_budget: true
   max_programs: 4
   max_run_times: 6
+  retries: true
 
 - device_id: "0006"
   code: ST8X_WF
@@ -40,13 +41,15 @@
   supports_water_budget: true
   max_programs: 4
   max_run_times: 6
+  retries: true
 
 - device_id: "0010"
   code: MOCK_ESP_ME2
   name: ESP=Me2
   supports_water_budget: true
   max_programs: 4
   max_run_times: 6
+  retries: true
 
 - device_id: "000a"
   code: ESP_TM2v2
@@ -75,6 +78,7 @@
   supports_water_budget: true
   max_programs: 4
   max_run_times: 6
+  retries: true
 
 - device_id: "0103"
   code: ESP_RZXe2

requirements_dev.txt

@@ -1,5 +1,6 @@
 -e .
 aiohttp==3.8.4
+aiohttp_retry==2.8.3
 freezegun==1.2.2
 mitmproxy==9.0.1
 parameterized==0.9.0

setup.cfg

@@ -21,6 +21,8 @@ install_requires =
   pydantic>=1.10.4
   python-dateutil>=2.8.2
   ical>=4.2.9
+  aiohttp_retry>=2.8.3
+
 install_package_data = True
 package_dir =
   = .

tests/test_async_client.py

@@ -133,6 +133,7 @@ async def test_get_model_and_version(
     assert model_info.supports_water_budget
     assert model_info.max_programs == 3
     assert model_info.max_run_times == 4
+    assert not model_info.retries
 
 
 async def test_get_available_stations(
@@ -145,6 +146,69 @@ async def test_get_available_stations(
     assert stations.active_set == {1, 2, 3, 4, 5, 6, 7}
 
 
+async def test_device_busy_retries(
+    rainbird_controller: Callable[[], Awaitable[AsyncRainbirdController]],
+    rainbird_client: Callable[[], Awaitable[AsyncRainbirdClient]],
+    api_response: Callable[[...], Awaitable[None]],
+    response: ResponseResult,
+) -> None:
+    """Test a basic request failure handling with retries."""
+    controller = await rainbird_controller()
+    api_response("82", modelID=0x09, protocolRevisionMajor=1, protocolRevisionMinor=3)
+    result = await controller.get_model_and_version()
+    assert result.model_code == "ESP_ME3"
+    assert result.model_info.retries
+
+    # Make two attempts then succeed
+    response(aiohttp.web.Response(status=503))
+    response(aiohttp.web.Response(status=503))
+    api_response("83", pageNumber=1, setStations=0x7F000000)
+
+    stations = await controller.get_available_stations()
+    assert stations.active_set == {1, 2, 3, 4, 5, 6, 7}
+
+
+async def test_non_retryable_errors(
+    rainbird_controller: Callable[[], Awaitable[AsyncRainbirdController]],
+    rainbird_client: Callable[[], Awaitable[AsyncRainbirdClient]],
+    api_response: Callable[[...], Awaitable[None]],
+    response: ResponseResult,
+) -> None:
+    """Test a basic request failure handling with retries."""
+    controller = await rainbird_controller()
+    api_response("82", modelID=0x09, protocolRevisionMajor=1, protocolRevisionMinor=3)
+    result = await controller.get_model_and_version()
+    assert result.model_code == "ESP_ME3"
+    assert result.model_info.retries
+
+    # Other types of errors are not retried
+    response(aiohttp.web.Response(status=403))
+    with pytest.raises(RainbirdAuthException):
+        await controller.get_available_stations()
+
+
+
+async def test_device_busy_retries_not_enabled(
+    rainbird_controller: Callable[[], Awaitable[AsyncRainbirdController]],
+    rainbird_client: Callable[[], Awaitable[AsyncRainbirdClient]],
+    api_response: Callable[[...], Awaitable[None]],
+    response: ResponseResult,
+) -> None:
+    """Test a basic request failure handling for device without retries."""
+    controller = await rainbird_controller()
+    api_response("82", modelID=0x0A, protocolRevisionMajor=1, protocolRevisionMinor=3)
+    result = await controller.get_model_and_version()
+    assert result == ModelAndVersion(0x0A, 1, 3)
+    assert result.model_code == "ESP_TM2v2"
+    assert result.model_name == "ESP-TM2"
+    assert not result.model_info.retries
+
+    response(aiohttp.web.Response(status=503))
+
+    with pytest.raises(RainbirdDeviceBusyException):
+        await controller.get_available_stations()
+
+
 async def test_get_serial_number(
     rainbird_controller: Callable[[], Awaitable[AsyncRainbirdController]],
     api_response: Callable[[...], Awaitable[None]],