From 647ea2878d2ccfbbe0d953a8a25ef3103b3026c0 Mon Sep 17 00:00:00 2001 From: Ashton Peters Date: Wed, 11 Nov 2026 08:51:22 +1300 Subject: [PATCH] fix: don't resend non-read commands over MQTT after a local timeout A V1 command that timed out on the local channel was retried over MQTT. By then the request had already been published, so the device may have received and run it, and the retry could run a write command (start, set, save map) twice. Only read-only commands (get_*, app_get_*) are now retried after a timeout; others raise RoborockTimeout. Failures before publish still fall back as before. Co-Authored-By: Claude Opus 5.5 --- roborock/devices/rpc/v1_channel.py | 20 ++++++++++++-- tests/devices/rpc/test_v1_channel.py | 41 +++++++++++++++++++++++++++- 2 files changed, 57 insertions(+), 4 deletions(-) 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,