diff --git a/roborock/devices/rpc/v1_channel.py b/roborock/devices/rpc/v1_channel.py index 8259994ad..75a5a566d 100644 --- a/roborock/devices/rpc/v1_channel.py +++ b/roborock/devices/rpc/v1_channel.py @@ -17,7 +17,7 @@ from roborock.devices.transport.channel import Channel from roborock.devices.transport.local_channel import LocalChannel, LocalSession, create_local_session from roborock.devices.transport.mqtt_channel import MqttChannel -from roborock.exceptions import RoborockException +from roborock.exceptions import RoborockException, RoborockTimeout from roborock.mqtt.health_manager import HealthManager from roborock.mqtt.session import MqttParams, MqttSession from roborock.protocols.v1_protocol import ( @@ -47,6 +47,12 @@ _T = TypeVar("_T", bound=RoborockBase) _TIMEOUT = 10.0 +_READ_ONLY_PREFIXES = ("get_", "app_get_") + + +def _is_read_only(method: CommandType) -> bool: + """Return True if the command only reads state, so it is safe to resend after a timeout.""" + return method.startswith(_READ_ONLY_PREFIXES) # Exponential backoff parameters for reconnecting to local @@ -89,10 +95,18 @@ async def send_command( request = RequestMessage(method, params=params) # Try each channel in order until one succeeds - last_exception = None + last_exception: RoborockException | None = None for strategy in self._rpc_strategies_cb(): try: decoded_response = await self._send_rpc(strategy, request, self._logger) + except RoborockTimeout as e: + # The request was published, so the device may have received and run it. Only + # read-only commands are safe to resend on another channel; resending anything + # else could make the device act twice. + if not _is_read_only(method): + raise + self._logger.debug("Command %s timed out on %s channel: %s", method, strategy.name, e) + last_exception = e except RoborockException as e: self._logger.debug("Command %s failed on %s channel: %s", method, strategy.name, e) last_exception = e @@ -154,7 +168,7 @@ def find_response(response_message: RoborockMessage) -> None: if strategy.health_manager: await strategy.health_manager.on_timeout() future.cancel() - raise RoborockException(f"Command timed out after {_TIMEOUT}s") from ex + raise RoborockTimeout(f"Command timed out after {_TIMEOUT}s") from ex finally: unsub() if strategy.health_manager: diff --git a/tests/devices/rpc/test_v1_channel.py b/tests/devices/rpc/test_v1_channel.py index c153c752d..c65ff5f5d 100644 --- a/tests/devices/rpc/test_v1_channel.py +++ b/tests/devices/rpc/test_v1_channel.py @@ -16,7 +16,7 @@ from roborock.devices.cache import DeviceCache, DeviceCacheData, InMemoryCache from roborock.devices.rpc.v1_channel import V1Channel from roborock.devices.transport.local_channel import LocalSession -from roborock.exceptions import RoborockException +from roborock.exceptions import RoborockException, RoborockTimeout from roborock.protocol import ( create_local_decoder, create_local_encoder, @@ -433,6 +433,45 @@ async def test_v1_channel_send_pick_first_available( assert result.state == RoborockStateCode.cleaning +@pytest.mark.parametrize( + ("command", "expect_mqtt_resend"), + [ + (RoborockCommand.GET_STATUS, True), + (RoborockCommand.APP_START, False), + ], + ids=["read-falls-back-to-mqtt", "write-not-resent-on-mqtt"], +) +async def test_v1_channel_send_command_local_timeout( + v1_channel: V1Channel, + rpc_channel: V1RpcChannel, + mock_mqtt_channel: Mock, + mock_local_channel: Mock, + command: RoborockCommand, + expect_mqtt_resend: bool, +) -> None: + """A command that times out locally is only resent over MQTT if it is read-only. + + After a timeout the request has already been published, so the device may have run it. + """ + mock_mqtt_channel.response_queue.append(TEST_NETWORK_INFO_RESPONSE) + await v1_channel.subscribe(Mock()) + mqtt_published_before = len(mock_mqtt_channel.published_messages) + + # Local publish succeeds but no response arrives; MQTT would answer. + mock_mqtt_channel.response_queue.append(TEST_RESPONSE) + with patch("roborock.devices.rpc.v1_channel._TIMEOUT", 0.01): + if expect_mqtt_resend: + result = await rpc_channel.send_command(command, response_type=StatusV2) + assert result.state == RoborockStateCode.cleaning + else: + with pytest.raises(RoborockTimeout): + await rpc_channel.send_command(command) + + mock_local_channel.publish.assert_called_once() + mqtt_resent = len(mock_mqtt_channel.published_messages) > mqtt_published_before + assert mqtt_resent == expect_mqtt_resend + + async def test_v1_channel_send_decoded_command_with_params( v1_channel: V1Channel, rpc_channel: V1RpcChannel,