From e61795a320aef5f1978122e1210e252ec5e9476d Mon Sep 17 00:00:00 2001 From: infy1995 Date: Fri, 13 Nov 2026 17:14:44 +0100 Subject: [PATCH 1/4] fix: restart the MQTT session when one device stops receiving messages The broker can stop delivering one device's messages while the shared session keeps working for the others. The health manager counted timeouts for the whole session and any successful response reset the count, so a working device hid a failing one and the session was never restarted (#1012). Subscribing to the topic again on the same connection does not help; only a new connection does. Track MQTT requests per device and restart the session when a device has 3 timeouts within 30 minutes. A single response in between doesn't reset the count, since a failing connection may still deliver some of them; the device counts as recovered after 3 responses in a row. Timeouts from several devices in a row still restart the session at once. A device that stays unreachable restarts the session less and less often: after 30 minutes, then 1, 2 and 4 hours, and at most every 6 hours until it responds again. Each health decision is logged at info level. --- roborock/devices/rpc/v1_channel.py | 4 +- roborock/devices/transport/mqtt_channel.py | 8 +- roborock/mqtt/health_manager.py | 225 ++++++++- roborock/testing/channel.py | 4 +- tests/devices/transport/test_mqtt_channel.py | 20 + tests/mqtt/test_health_manager.py | 501 ++++++++++++++++++- 6 files changed, 728 insertions(+), 34 deletions(-) diff --git a/roborock/devices/rpc/v1_channel.py b/roborock/devices/rpc/v1_channel.py index 8259994ad..2d7d6e6c2 100644 --- a/roborock/devices/rpc/v1_channel.py +++ b/roborock/devices/rpc/v1_channel.py @@ -18,7 +18,7 @@ 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.mqtt.health_manager import HealthManager +from roborock.mqtt.health_manager import DeviceHealthManager from roborock.mqtt.session import MqttParams, MqttSession from roborock.protocols.v1_protocol import ( CommandType, @@ -67,7 +67,7 @@ class RpcStrategy: channel: LocalChannel | MqttChannel encoder: Callable[[RequestMessage], RoborockMessage] decoder: Callable[[RoborockMessage], ResponseMessage | MapResponse | None] - health_manager: HealthManager | None = None + health_manager: DeviceHealthManager | None = None class RpcChannel(V1RpcChannel): diff --git a/roborock/devices/transport/mqtt_channel.py b/roborock/devices/transport/mqtt_channel.py index 1a0f81e64..0c5156723 100644 --- a/roborock/devices/transport/mqtt_channel.py +++ b/roborock/devices/transport/mqtt_channel.py @@ -7,7 +7,7 @@ from roborock.callbacks import decoder_callback from roborock.data import HomeDataDevice, RRiot, UserData from roborock.exceptions import RoborockException -from roborock.mqtt.health_manager import HealthManager +from roborock.mqtt.health_manager import DeviceHealthManager from roborock.mqtt.session import MqttParams, MqttQos, MqttSession, MqttSessionException from roborock.protocol import create_mqtt_decoder, create_mqtt_encoder from roborock.roborock_message import RoborockMessage @@ -45,9 +45,9 @@ def is_connected(self) -> bool: return self._mqtt_session.connected @property - def health_manager(self) -> HealthManager: - """Return the health manager for the session.""" - return self._mqtt_session.health_manager + def health_manager(self) -> DeviceHealthManager: + """Return the health manager that reports this device's events to the session.""" + return DeviceHealthManager(self._mqtt_session.health_manager, self._duid) @property def is_local_connected(self) -> bool: diff --git a/roborock/mqtt/health_manager.py b/roborock/mqtt/health_manager.py index 9e5a23ce6..f977ae9d3 100644 --- a/roborock/mqtt/health_manager.py +++ b/roborock/mqtt/health_manager.py @@ -1,28 +1,90 @@ """A health manager for monitoring MQTT connections to Roborock devices. We observe a problem where sometimes the MQTT connection appears to be alive but -no messages are being received. To mitigate this, we track consecutive timeouts -and restart the connection if too many timeouts occur in succession. +no messages are being received. To mitigate this, we track timeouts and restart +the connection if too many of them occur. + +The broker can also stop delivering messages for a single device's topic while +the rest of the session keeps working, so timeouts are tracked both for the +session as a whole and for each device. Subscribing to the device's topic again +on the same connection does not bring the messages back; only a new connection +does, so the session is restarted. Such a failure can be partial, with some +requests still answered in between, so a device's timeouts are counted within a +time window rather than in a row. """ import datetime import logging +from collections import deque from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from typing import Any + +from roborock.diagnostics import redact_device_uid _LOGGER = logging.getLogger(__name__) -# Number of consecutive timeouts before considering the connection unhealthy. +# Number of timeouts before considering the connection unhealthy: in a row across the +# session, or within TIMEOUT_WINDOW for a single device. TIMEOUT_THRESHOLD = 3 +# Timeouts of a device older than this no longer count towards a restart. Successful +# requests in between don't reset the count, since a failing connection may still +# deliver some responses. +TIMEOUT_WINDOW = datetime.timedelta(minutes=30) + +# A failing device counts as recovered after this many successful requests in a row, or at +# its first successful request once its timeouts fell out of TIMEOUT_WINDOW. +RECOVERY_THRESHOLD = 3 + # We won't restart the session more often than this interval. RESTART_COOLDOWN = datetime.timedelta(minutes=30) +# A device that fails while the rest of the session works may simply be offline, +# which a restart can't fix. Until it responds again, the interval between the +# session restarts it triggers doubles after each restart, up to this maximum. +MAX_DEVICE_RESTART_COOLDOWN = datetime.timedelta(hours=6) + +# Number of health changes kept per device for diagnostics. +MAX_DEVICE_EVENTS = 10 + + +@dataclass +class _DeviceHealth: + """The MQTT request history of a single device.""" + + recent_timeouts: deque[datetime.datetime] = field(default_factory=deque) + successes_in_a_row: int = 0 + failing_since: datetime.datetime | None = None + restart_cooldown_start: datetime.datetime | None = None + restart_cooldown: datetime.timedelta | None = None + timeouts: int = 0 + successes: int = 0 + last_success: datetime.datetime | None = None + last_timeout: datetime.datetime | None = None + events: deque[dict[str, str]] = field(default_factory=lambda: deque(maxlen=MAX_DEVICE_EVENTS)) + + def add_event(self, time: datetime.datetime, event: str, **details: str) -> None: + """Record a change of the device's health for diagnostics.""" + self.events.append({"time": time.isoformat(), "event": event, **details}) + + def forget_old_timeouts(self, now: datetime.datetime) -> None: + """Drop the timeouts that fell out of the time window.""" + while self.recent_timeouts and now - self.recent_timeouts[0] >= TIMEOUT_WINDOW: + self.recent_timeouts.popleft() + + +def _isoformat(value: datetime.datetime | None) -> str | None: + """Format an optional timestamp for diagnostics.""" + return value.isoformat() if value else None + class HealthManager: """Manager for monitoring the health of MQTT connections. This tracks communication timeouts and can trigger restarts of the MQTT - session if too many timeouts occur in succession. + session if too many timeouts occur, either in a row across the whole + session or within a time window for a single device. """ def __init__(self, restart: Callable[[], Awaitable[None]]) -> None: @@ -32,29 +94,154 @@ def __init__(self, restart: Callable[[], Awaitable[None]]) -> None: restart: A callable to restart the MQTT session. """ self._consecutive_timeouts = 0 + self._timed_out_devices: set[str] = set() + self._devices: dict[str, _DeviceHealth] = {} self._restart = restart self._last_restart: datetime.datetime | None = None - async def on_success(self) -> None: - """Record a successful communication event.""" + def _device(self, device_id: str) -> _DeviceHealth: + """Return the history of a device, creating it on first use.""" + return self._devices.setdefault(device_id, _DeviceHealth()) + + def on_connected(self) -> None: + """Forget the timeouts counted on a previous connection. + + Timeouts on an old connection say nothing about a new one, whether the + session reconnected on its own or was restarted. + """ self._consecutive_timeouts = 0 + self._timed_out_devices.clear() + for device in self._devices.values(): + device.recent_timeouts.clear() - async def on_timeout(self) -> None: - """Record a timeout event. + async def on_success(self, device_id: str) -> None: + """Record a successful communication event for a device.""" + self._consecutive_timeouts = 0 + self._timed_out_devices.clear() + device = self._device(device_id) + now = datetime.datetime.now(datetime.UTC) + device.successes += 1 + device.successes_in_a_row += 1 + device.last_success = now + device.forget_old_timeouts(now) + if device.failing_since is None: + return + # A single response in between timeouts doesn't mean the device recovered yet. + if device.recent_timeouts and device.successes_in_a_row < RECOVERY_THRESHOLD: + return + _LOGGER.info( + "Device %s responds over MQTT again after timing out since %s", + redact_device_uid(device_id), + device.failing_since.isoformat(timespec="seconds"), + ) + device.add_event(now, "recovered") + device.recent_timeouts.clear() + device.failing_since = None + device.restart_cooldown_start = None + device.restart_cooldown = None + + async def on_timeout(self, device_id: str) -> None: + """Record a timeout event for a device. This may trigger a restart of the MQTT session if too many timeouts - have occurred in succession. + have occurred. """ + now = datetime.datetime.now(datetime.UTC) self._consecutive_timeouts += 1 - if self._consecutive_timeouts >= TIMEOUT_THRESHOLD: - now = datetime.datetime.now(datetime.UTC) - since_last = (now - self._last_restart) if self._last_restart else None - if since_last is None or since_last >= RESTART_COOLDOWN: + self._timed_out_devices.add(device_id) + device = self._device(device_id) + device_name = redact_device_uid(device_id) + if device.failing_since is None: + device.failing_since = now + _LOGGER.info("Device %s started timing out over MQTT", device_name) + device.add_event(now, "timeouts_started") + device.forget_old_timeouts(now) + device.recent_timeouts.append(now) + device.successes_in_a_row = 0 + device.timeouts += 1 + device.last_timeout = now + + # Only timeouts from several devices implicate the session itself. A single failing device + # may just be offline, even when it is the only device reporting to this session. + session_unhealthy = self._consecutive_timeouts >= TIMEOUT_THRESHOLD and len(self._timed_out_devices) > 1 + if not session_unhealthy and len(device.recent_timeouts) < TIMEOUT_THRESHOLD: + return + + since_last = (now - self._last_restart) if self._last_restart else None + if since_last is not None and since_last < RESTART_COOLDOWN: + _LOGGER.debug("Not restarting MQTT session for device %s, last restart was %s ago", device_name, since_last) + return + + reason = f"{self._consecutive_timeouts} timeouts in a row across {len(self._timed_out_devices)} devices" + if not session_unhealthy: + cooldown_start, cooldown = device.restart_cooldown_start, device.restart_cooldown + if cooldown_start is not None and cooldown is not None and now - cooldown_start < cooldown: _LOGGER.debug( - "Restarting MQTT session after %d consecutive timeouts (duration since last restart %s)", - self._consecutive_timeouts, - since_last, + "Not restarting MQTT session for device %s, it is still failing and may restart it again after %s", + device_name, + (cooldown_start + cooldown).isoformat(timespec="seconds"), ) - await self._restart() - self._last_restart = now - self._consecutive_timeouts = 0 + return + device.restart_cooldown_start = now + device.restart_cooldown = ( + min(cooldown * 2, MAX_DEVICE_RESTART_COOLDOWN) if cooldown is not None else RESTART_COOLDOWN + ) + reason = ( + f"device {device_name} timed out {len(device.recent_timeouts)} times in {TIMEOUT_WINDOW}; " + f"it may restart the session again in {device.restart_cooldown} if it keeps failing" + ) + + _LOGGER.info( + "Restarting MQTT session: %s (last restart: %s)", + reason, + f"{since_last} ago" if since_last is not None else "never", + ) + # Update the state before awaiting, so timeouts arriving meanwhile don't restart again. + self._last_restart = now + # The restart affects every device, so each one records it along with what caused it. + several_devices = len(self._timed_out_devices) > 1 + for tracked_id, tracked_device in self._devices.items(): + if several_devices: + cause = "several_devices" + else: + cause = "this_device" if tracked_id == device_id else "other_device" + tracked_device.add_event(now, "session_restart", cause=cause) + # Timeouts on the old connection say nothing about the new one. + tracked_device.recent_timeouts.clear() + self._consecutive_timeouts = 0 + self._timed_out_devices.clear() + await self._restart() + + def device_diagnostic_data(self, device_id: str) -> dict[str, Any]: + """Return diagnostics about a device's MQTT requests.""" + device = self._devices.get(device_id, _DeviceHealth()) + return { + "failing_since": _isoformat(device.failing_since), + "last_success": _isoformat(device.last_success), + "last_timeout": _isoformat(device.last_timeout), + "successes": device.successes, + "timeouts": device.timeouts, + # Newest first, so the current state of the device is at the top. + "events": [dict(event) for event in reversed(device.events)], + } + + +class DeviceHealthManager: + """Reports the outcome of a single device's MQTT requests to the session's health manager.""" + + def __init__(self, health_manager: HealthManager, device_id: str) -> None: + """Initialize the device health manager.""" + self._health_manager = health_manager + self._device_id = device_id + + async def on_success(self) -> None: + """Record a successful communication event.""" + await self._health_manager.on_success(self._device_id) + + async def on_timeout(self) -> None: + """Record a timeout event.""" + await self._health_manager.on_timeout(self._device_id) + + def diagnostic_data(self) -> dict[str, Any]: + """Return diagnostics about this device's MQTT requests.""" + return self._health_manager.device_diagnostic_data(self._device_id) diff --git a/roborock/testing/channel.py b/roborock/testing/channel.py index 65a36bc79..3e9f4f7d1 100644 --- a/roborock/testing/channel.py +++ b/roborock/testing/channel.py @@ -11,7 +11,7 @@ from unittest.mock import AsyncMock, MagicMock from roborock.devices.transport.channel import Channel -from roborock.mqtt.health_manager import HealthManager +from roborock.mqtt.health_manager import DeviceHealthManager, HealthManager from roborock.mqtt.session import MqttQos from roborock.protocols.v1_protocol import LocalProtocolVersion from roborock.roborock_message import RoborockMessage @@ -86,7 +86,7 @@ def __init__(self, is_local: bool = False): self.protocol_version = LocalProtocolVersion.V1 self.restart = AsyncMock() - self.health_manager = HealthManager(self.restart) + self.health_manager = DeviceHealthManager(HealthManager(self.restart), "fake-device") async def _connect(self) -> None: self._is_connected = True diff --git a/tests/devices/transport/test_mqtt_channel.py b/tests/devices/transport/test_mqtt_channel.py index 55e7b8bbb..dffde3fd1 100644 --- a/tests/devices/transport/test_mqtt_channel.py +++ b/tests/devices/transport/test_mqtt_channel.py @@ -10,6 +10,7 @@ from roborock.data import HomeData, UserData from roborock.devices.transport.mqtt_channel import MqttChannel +from roborock.mqtt.health_manager import HealthManager from roborock.mqtt.session import MqttParams from roborock.protocol import create_mqtt_decoder, create_mqtt_encoder from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol @@ -268,3 +269,22 @@ def failing_callback(message: RoborockMessage) -> None: # Unsubscribe all remaining subscribers unsub1() unsub2() + + +async def test_health_is_tracked_per_device(mqtt_session: Mock) -> None: + """Test that a working device on the session does not hide a failing device's timeouts.""" + restart = AsyncMock() + mqtt_session.health_manager = HealthManager(restart) + failing, working = ( + MqttChannel( + mqtt_session, duid=duid, local_key=TEST_LOCAL_KEY, rriot=USER_DATA.rriot, mqtt_params=TEST_MQTT_PARAMS + ) + for duid in ("failing", "working") + ) + + for _ in range(3): + restart.assert_not_called() + await working.health_manager.on_success() + await failing.health_manager.on_timeout() + + restart.assert_called_once() diff --git a/tests/mqtt/test_health_manager.py b/tests/mqtt/test_health_manager.py index 94b253b4b..86a6ee5d2 100644 --- a/tests/mqtt/test_health_manager.py +++ b/tests/mqtt/test_health_manager.py @@ -1,15 +1,28 @@ """Tests for the health manager.""" +import asyncio import datetime +import logging from unittest.mock import AsyncMock, patch -from roborock.mqtt.health_manager import HealthManager +import pytest + +from roborock.mqtt.health_manager import ( + MAX_DEVICE_EVENTS, + MAX_DEVICE_RESTART_COOLDOWN, + RECOVERY_THRESHOLD, + TIMEOUT_WINDOW, + DeviceHealthManager, + HealthManager, +) + +DEVICE_ID = "device-1" async def test_health_manager_restart_called_after_timeouts() -> None: """Test that the health manager calls restart after consecutive timeouts.""" restart = AsyncMock() - health_manager = HealthManager(restart=restart) + health_manager = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID) await health_manager.on_timeout() await health_manager.on_timeout() @@ -19,29 +32,92 @@ async def test_health_manager_restart_called_after_timeouts() -> None: restart.assert_called_once() -async def test_health_manager_success_resets_counter() -> None: - """Test that a successful message resets the timeout counter.""" +async def test_responses_in_between_do_not_hide_a_partial_failure() -> None: + """Test that timeouts interleaved with successful responses still restart the session.""" restart = AsyncMock() - health_manager = HealthManager(restart=restart) + health_manager = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID) await health_manager.on_timeout() + await health_manager.on_success() await health_manager.on_timeout() + await health_manager.on_success() restart.assert_not_called() + await health_manager.on_timeout() + restart.assert_called_once() + # The device kept timing out, so the responses in between are not recorded as a recovery. + assert [event["event"] for event in health_manager.diagnostic_data()["events"]] == [ + "session_restart", + "timeouts_started", + ] + + +async def test_responses_in_a_row_mark_recovery() -> None: + """Test that a device answering several requests in a row recovers before its timeouts expire.""" + restart = AsyncMock() + health_manager = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID) + + await health_manager.on_timeout() + await health_manager.on_timeout() + for _ in range(RECOVERY_THRESHOLD - 1): + await health_manager.on_success() + assert health_manager.diagnostic_data()["failing_since"] is not None + await health_manager.on_success() + assert health_manager.diagnostic_data()["failing_since"] is None + assert [event["event"] for event in health_manager.diagnostic_data()["events"]] == [ + "recovered", + "timeouts_started", + ] + # The timeouts before the recovery no longer count towards a restart. await health_manager.on_timeout() await health_manager.on_timeout() restart.assert_not_called() - await health_manager.on_timeout() + +async def test_timeouts_before_a_reconnect_do_not_count() -> None: + """Test that a new connection is not restarted because of the old connection's timeouts.""" + restart = AsyncMock() + health_manager = HealthManager(restart=restart) + device = DeviceHealthManager(health_manager, DEVICE_ID) + + await device.on_timeout() + await device.on_timeout() + health_manager.on_connected() + + await device.on_timeout() + await device.on_timeout() + restart.assert_not_called() + + await device.on_timeout() restart.assert_called_once() +async def test_old_timeouts_do_not_count() -> None: + """Test that only the timeouts within the time window restart the session.""" + restart = AsyncMock() + health_manager = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID) + + with patch("roborock.mqtt.health_manager.datetime") as mock_datetime: + now = datetime.datetime(2023, 1, 1, 12, 0, 0) + mock_datetime.datetime.now.return_value = now + await health_manager.on_timeout() + await health_manager.on_timeout() + + mock_datetime.datetime.now.return_value = now + TIMEOUT_WINDOW + await health_manager.on_timeout() + await health_manager.on_timeout() + restart.assert_not_called() + + await health_manager.on_timeout() + restart.assert_called_once() + + async def test_cooldown() -> None: """Test that the health manager respects the restart cooldown.""" restart = AsyncMock() - health_manager = HealthManager(restart=restart) + health_manager = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID) with patch("roborock.mqtt.health_manager.datetime") as mock_datetime: now = datetime.datetime(2023, 1, 1, 12, 0, 0) @@ -71,3 +147,414 @@ async def test_cooldown() -> None: await health_manager.on_timeout() await health_manager.on_timeout() restart.assert_called_once() + + +async def test_timeouts_across_devices_restart_session() -> None: + """Test that timeouts spread across devices still count towards the session.""" + restart = AsyncMock() + health_manager = HealthManager(restart=restart) + + await DeviceHealthManager(health_manager, "device-1").on_timeout() + await DeviceHealthManager(health_manager, "device-2").on_timeout() + restart.assert_not_called() + + await DeviceHealthManager(health_manager, "device-3").on_timeout() + restart.assert_called_once() + + +async def test_other_device_success_does_not_hide_failing_device() -> None: + """Test that one device's timeouts restart the session while another device works.""" + restart = AsyncMock() + health_manager = HealthManager(restart=restart) + failing = DeviceHealthManager(health_manager, "failing") + working = DeviceHealthManager(health_manager, "working") + + await failing.on_timeout() + await failing.on_timeout() + await working.on_success() + restart.assert_not_called() + + await failing.on_timeout() + restart.assert_called_once() + + +async def test_device_needs_timeouts_of_its_own() -> None: + """Test that timeouts of other devices don't count towards a device's restart.""" + restart = AsyncMock() + health_manager = HealthManager(restart=restart) + device = DeviceHealthManager(health_manager, DEVICE_ID) + other = DeviceHealthManager(health_manager, "other") + + await device.on_timeout() + await device.on_success() + await other.on_timeout() + await other.on_success() + await device.on_timeout() + restart.assert_not_called() + + +async def test_restart_clears_counters_of_all_devices() -> None: + """Test that every device starts counting again after a session restart.""" + restart = AsyncMock() + health_manager = HealthManager(restart=restart) + first = DeviceHealthManager(health_manager, "first") + second = DeviceHealthManager(health_manager, "second") + working = DeviceHealthManager(health_manager, "working") + + with patch("roborock.mqtt.health_manager.datetime") as mock_datetime: + now = datetime.datetime(2023, 1, 1, 12, 0, 0) + mock_datetime.datetime.now.return_value = now + + await second.on_timeout() + await second.on_timeout() + await working.on_success() + await first.on_timeout() + await first.on_timeout() + await working.on_success() + await first.on_timeout() + restart.assert_called_once() + restart.reset_mock() + + # The restart gave "second" a new subscription, so its earlier timeouts no longer count. + mock_datetime.datetime.now.return_value = now + datetime.timedelta(minutes=31) + await working.on_success() + await second.on_timeout() + restart.assert_not_called() + + +async def test_single_failing_device_cooldown_grows() -> None: + """Test that a device that stays unreachable restarts a working session less and less often.""" + restart = AsyncMock() + health_manager = HealthManager(restart=restart) + failing = DeviceHealthManager(health_manager, "failing") + working = DeviceHealthManager(health_manager, "working") + + async def time_out_while_session_works() -> None: + for _ in range(3): + await working.on_success() + await failing.on_timeout() + + with patch("roborock.mqtt.health_manager.datetime") as mock_datetime: + now = datetime.datetime(2023, 1, 1, 12, 0, 0) + mock_datetime.datetime.now.return_value = now + await time_out_while_session_works() + restart.assert_called_once() + + # The interval between restarts doubles from the session cooldown up to the maximum. + hours = datetime.timedelta(hours=1) + for cooldown in ( + hours / 2, + hours, + 2 * hours, + 4 * hours, + MAX_DEVICE_RESTART_COOLDOWN, + MAX_DEVICE_RESTART_COOLDOWN, + ): + restart.reset_mock() + mock_datetime.datetime.now.return_value = now + cooldown - datetime.timedelta(minutes=1) + await time_out_while_session_works() + restart.assert_not_called() + + now += cooldown + mock_datetime.datetime.now.return_value = now + await time_out_while_session_works() + restart.assert_called_once() + + +async def test_device_success_clears_device_cooldown() -> None: + """Test that a device that recovered can restart the session again if it fails later.""" + restart = AsyncMock() + health_manager = HealthManager(restart=restart) + failing = DeviceHealthManager(health_manager, "failing") + working = DeviceHealthManager(health_manager, "working") + + async def time_out_while_session_works() -> None: + for _ in range(3): + await working.on_success() + await failing.on_timeout() + + with patch("roborock.mqtt.health_manager.datetime") as mock_datetime: + now = datetime.datetime(2023, 1, 1, 12, 0, 0) + mock_datetime.datetime.now.return_value = now + + await time_out_while_session_works() + restart.assert_called_once() + restart.reset_mock() + + await failing.on_success() + + # Only the session cooldown applies now. + mock_datetime.datetime.now.return_value = now + datetime.timedelta(minutes=31) + await time_out_while_session_works() + restart.assert_called_once() + + +async def test_device_diagnostics_are_separate_per_device() -> None: + """Test that diagnostics keep each device's history apart and survive counter resets.""" + restart = AsyncMock() + health_manager = HealthManager(restart=restart) + failing = DeviceHealthManager(health_manager, "failing") + working = DeviceHealthManager(health_manager, "working") + + with patch("roborock.mqtt.health_manager.datetime") as mock_datetime: + now = datetime.datetime(2023, 1, 1, 12, 0, 0, tzinfo=datetime.UTC) + mock_datetime.datetime.now.return_value = now + await failing.on_success() + + mock_datetime.datetime.now.return_value = now + datetime.timedelta(hours=1) + for _ in range(4): + await working.on_success() + await failing.on_timeout() + + last_timeout = "2023-01-01T13:00:00+00:00" + assert failing.diagnostic_data() == { + "failing_since": last_timeout, + "last_success": "2023-01-01T12:00:00+00:00", + "last_timeout": last_timeout, + "successes": 1, + "timeouts": 4, + "events": [ + {"time": last_timeout, "event": "session_restart", "cause": "this_device"}, + {"time": last_timeout, "event": "timeouts_started"}, + ], + } + assert working.diagnostic_data() == { + "failing_since": None, + "last_success": last_timeout, + "last_timeout": None, + "successes": 4, + "timeouts": 0, + "events": [ + {"time": last_timeout, "event": "session_restart", "cause": "other_device"}, + ], + } + + +async def test_diagnostics_for_device_without_requests() -> None: + """Test the diagnostics of a device that has not used MQTT yet.""" + device = DeviceHealthManager(HealthManager(restart=AsyncMock()), DEVICE_ID) + + assert device.diagnostic_data() == { + "failing_since": None, + "last_success": None, + "last_timeout": None, + "successes": 0, + "timeouts": 0, + "events": [], + } + + +async def test_diagnostics_record_when_timeouts_start_and_recover() -> None: + """Test that only changes of a device's health are recorded.""" + device = DeviceHealthManager(HealthManager(restart=AsyncMock()), DEVICE_ID) + + with patch("roborock.mqtt.health_manager.datetime") as mock_datetime: + start = datetime.datetime(2023, 1, 1, 12, 0, 0, tzinfo=datetime.UTC) + mock_datetime.datetime.now.return_value = start + await device.on_success() + await device.on_timeout() + await device.on_timeout() + + # A response while the timeouts still count is not a recovery yet. + mock_datetime.datetime.now.return_value = start + datetime.timedelta(minutes=10) + await device.on_success() + + recovered = start + TIMEOUT_WINDOW + mock_datetime.datetime.now.return_value = recovered + await device.on_success() + await device.on_success() + + assert device.diagnostic_data()["events"] == [ + {"time": recovered.isoformat(), "event": "recovered"}, + {"time": start.isoformat(), "event": "timeouts_started"}, + ] + assert device.diagnostic_data()["failing_since"] is None + + +async def test_diagnostics_record_recovery_after_restart() -> None: + """Test that a device that recovers after a session restart is recorded as recovered.""" + restart = AsyncMock() + device = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID) + + with patch("roborock.mqtt.health_manager.datetime") as mock_datetime: + start = datetime.datetime(2023, 1, 1, 12, 0, 0, tzinfo=datetime.UTC) + mock_datetime.datetime.now.return_value = start + for _ in range(3): + await device.on_timeout() + restart.assert_called_once() + + recovered = start + datetime.timedelta(minutes=10) + mock_datetime.datetime.now.return_value = recovered + await device.on_success() + + assert device.diagnostic_data()["events"] == [ + {"time": recovered.isoformat(), "event": "recovered"}, + {"time": start.isoformat(), "event": "session_restart", "cause": "this_device"}, + {"time": start.isoformat(), "event": "timeouts_started"}, + ] + assert device.diagnostic_data()["failing_since"] is None + + +async def test_health_changes_are_logged(caplog: pytest.LogCaptureFixture) -> None: + """Test that every decision about a failing device is visible in the info log.""" + caplog.set_level(logging.INFO, logger="roborock.mqtt.health_manager") + health_manager = HealthManager(restart=AsyncMock()) + failing = DeviceHealthManager(health_manager, "failing") + working = DeviceHealthManager(health_manager, "working") + + for _ in range(3): + await working.on_success() + await failing.on_timeout() + await failing.on_success() + + messages = [record.getMessage() for record in caplog.records] + assert [message.split(" ")[0] for message in messages] == ["Device", "Restarting", "Device"] + assert "started timing out over MQTT" in messages[0] + assert "may restart the session again in 0:30:00" in messages[1] + assert "(last restart: never)" in messages[1] + assert "responds over MQTT again" in messages[2] + + +async def test_only_known_device_is_paced_like_any_failing_device() -> None: + """Test that the restarts triggered by the only device of a session become rarer.""" + restart = AsyncMock() + device = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID) + + with patch("roborock.mqtt.health_manager.datetime") as mock_datetime: + now = datetime.datetime(2023, 1, 1, 12, 0, 0) + mock_datetime.datetime.now.return_value = now + for _ in range(3): + await device.on_timeout() + restart.assert_called_once() + restart.reset_mock() + + # The second restart is allowed after 30 minutes, the third only after another hour. + now += datetime.timedelta(minutes=31) + mock_datetime.datetime.now.return_value = now + for _ in range(3): + await device.on_timeout() + restart.assert_called_once() + restart.reset_mock() + + mock_datetime.datetime.now.return_value = now + datetime.timedelta(minutes=45) + for _ in range(3): + await device.on_timeout() + restart.assert_not_called() + + +async def test_concurrent_timeouts_restart_once() -> None: + """Test that timeouts arriving while the session is restarting don't restart it again.""" + restarted = asyncio.Event() + calls = 0 + + async def slow_restart() -> None: + nonlocal calls + calls += 1 + await restarted.wait() + + health_manager = HealthManager(restart=slow_restart) + failing = DeviceHealthManager(health_manager, "failing") + await DeviceHealthManager(health_manager, "working").on_success() + + await failing.on_timeout() + await failing.on_timeout() + pending = [asyncio.create_task(failing.on_timeout()) for _ in range(5)] + await asyncio.sleep(0) + restarted.set() + await asyncio.gather(*pending) + + assert calls == 1 + assert [event["event"] for event in failing.diagnostic_data()["events"]] == [ + "session_restart", + "timeouts_started", + ] + + +async def test_session_failure_is_recorded_on_every_device() -> None: + """Test that a restart caused by several devices names them all as the cause.""" + restart = AsyncMock() + health_manager = HealthManager(restart=restart) + first = DeviceHealthManager(health_manager, "first") + second = DeviceHealthManager(health_manager, "second") + + await first.on_timeout() + await second.on_timeout() + await first.on_timeout() + + restart.assert_called_once() + for device in (first, second): + assert device.diagnostic_data()["events"][0]["cause"] == "several_devices" + + +async def test_diagnostics_keep_only_recent_events() -> None: + """Test that the event history of a device is bounded.""" + device = DeviceHealthManager(HealthManager(restart=AsyncMock()), DEVICE_ID) + + with patch("roborock.mqtt.health_manager.datetime") as mock_datetime: + now = datetime.datetime(2023, 1, 1, 12, 0, 0) + for _ in range(MAX_DEVICE_EVENTS): + mock_datetime.datetime.now.return_value = now + await device.on_timeout() + now += TIMEOUT_WINDOW + mock_datetime.datetime.now.return_value = now + await device.on_success() + + assert len(device.diagnostic_data()["events"]) == MAX_DEVICE_EVENTS + + +async def test_device_cooldown_applies_without_successes_from_other_devices() -> None: + """Test that a failing device keeps its cooldown when other devices report nothing in between.""" + restart = AsyncMock() + health_manager = HealthManager(restart=restart) + failing = DeviceHealthManager(health_manager, "failing") + working = DeviceHealthManager(health_manager, "working") + + with patch("roborock.mqtt.health_manager.datetime") as mock_datetime: + now = datetime.datetime(2023, 1, 1, 12, 0, 0) + mock_datetime.datetime.now.return_value = now + await working.on_success() + + for _ in range(3): + await failing.on_timeout() + restart.assert_called_once() + restart.reset_mock() + + now += datetime.timedelta(minutes=31) + mock_datetime.datetime.now.return_value = now + for _ in range(3): + await failing.on_timeout() + restart.assert_called_once() + restart.reset_mock() + + # Past the session cooldown, but only this device is failing and its own cooldown has grown. + mock_datetime.datetime.now.return_value = now + datetime.timedelta(minutes=45) + for _ in range(3): + await failing.on_timeout() + restart.assert_not_called() + + +async def test_timeouts_from_several_devices_bypass_device_cooldown() -> None: + """Test that a session failing for several devices restarts during one device's cooldown.""" + restart = AsyncMock() + health_manager = HealthManager(restart=restart) + first = DeviceHealthManager(health_manager, "first") + second = DeviceHealthManager(health_manager, "second") + + with patch("roborock.mqtt.health_manager.datetime") as mock_datetime: + now = datetime.datetime(2023, 1, 1, 12, 0, 0) + mock_datetime.datetime.now.return_value = now + await second.on_success() + + for _ in range(3): + await first.on_timeout() + restart.assert_called_once() + restart.reset_mock() + + # The second device is failing too, so the session is restarted again. + mock_datetime.datetime.now.return_value = now + datetime.timedelta(hours=1) + await first.on_timeout() + await second.on_timeout() + restart.assert_not_called() + + await first.on_timeout() + restart.assert_called_once() From 61f052c15c144e6e9893f8d647e93a24e1de982e Mon Sep 17 00:00:00 2001 From: infy1995 Date: Fri, 13 Nov 2026 17:16:10 +0100 Subject: [PATCH 2/4] fix: reconnect right away after a forced MQTT session restart A restart requested by the health manager disconnects a session whose broker was reachable, so waiting for the backoff interval only kept every device on the session without MQTT for ten seconds. Reconnect at once after such a restart; connections lost for any other reason, and a failed first reconnect attempt, still wait for the backoff. Every new connection also clears the timeouts counted on the previous one. --- roborock/mqtt/roborock_session.py | 16 +++++-- tests/mqtt/test_roborock_session.py | 73 +++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 3 deletions(-) diff --git a/roborock/mqtt/roborock_session.py b/roborock/mqtt/roborock_session.py index a64992718..9abe27f49 100644 --- a/roborock/mqtt/roborock_session.py +++ b/roborock/mqtt/roborock_session.py @@ -79,6 +79,7 @@ def __init__( self._client_lock = asyncio.Lock() self._listeners: CallbackMap[str, bytes] = CallbackMap(_LOGGER) self._connection_task: asyncio.Task[None] | None = None + self._restart_requested = False self._topic_idle_timeout = topic_idle_timeout self._idle_timers: dict[str, asyncio.Task[None]] = {} self._diagnostics = params.diagnostics @@ -145,12 +146,15 @@ async def close(self) -> None: async def restart(self) -> None: """Force the session to disconnect and reconnect. - The active connection task will be cancelled and restarted in the background, retried by - the reconnect loop. This is a no-op if there is no active connection. + The active connection task will be cancelled and restarted in the background by the + reconnect loop. The broker was reachable, so the first reconnect attempt is made right + away instead of after the backoff interval. This is a no-op if there is no active + connection. """ _LOGGER.info("Forcing MQTT session restart") self._diagnostics.increment("restart") - if self._connection_task: + if self._connection_task and not self._connection_task.done(): + self._restart_requested = True self._connection_task.cancel() else: _LOGGER.debug("No message loop task to cancel") @@ -175,6 +179,7 @@ async def _run_reconnect_loop(self, start_future: asyncio.Future[None] | None) - self._healthy = False start_future = None + restart_requested, self._restart_requested = self._restart_requested, False if self._stop: _LOGGER.debug("MQTT session closed, stopping retry loop") return @@ -188,6 +193,10 @@ async def _run_reconnect_loop(self, start_future: asyncio.Future[None] | None) - return self._backoff = MIN_BACKOFF_INTERVAL continue + if restart_requested: + _LOGGER.info("MQTT session disconnected for a restart, reconnecting now") + self._diagnostics.increment("reconnect_restart") + continue _LOGGER.info("MQTT session disconnected, retrying in %s seconds", self._backoff.total_seconds()) self._diagnostics.increment("reconnect_wait") await asyncio.sleep(self._backoff.total_seconds()) @@ -205,6 +214,7 @@ async def _run_connection(self, start_future: asyncio.Future[None] | None) -> No async with self._mqtt_client(self._params) as client: self._backoff = MIN_BACKOFF_INTERVAL self._healthy = True + self._health_manager.on_connected() _LOGGER.info("MQTT Session connected.") if start_future and not start_future.done(): start_future.set_result(None) diff --git a/tests/mqtt/test_roborock_session.py b/tests/mqtt/test_roborock_session.py index 65ccbbdf5..4206a1c67 100644 --- a/tests/mqtt/test_roborock_session.py +++ b/tests/mqtt/test_roborock_session.py @@ -539,6 +539,79 @@ def succeed_then_fail_unauthorized() -> Any: await session.close() +async def test_restart_reconnects_without_backoff( + mock_aenter_client: AsyncMock, + mqtt_client_lite: AsyncMock, +) -> None: + """Test that a forced restart reconnects right away instead of waiting for the backoff.""" + params = copy.deepcopy(FAKE_PARAMS) + params.diagnostics = Diagnostics() + with patch("roborock.mqtt.roborock_session.MIN_BACKOFF_INTERVAL", datetime.timedelta(hours=1)): + session = await create_mqtt_session(params) + await session.subscribe("topic-1", Subscriber().append) + + try: + await session.restart() + async with asyncio.timeout(1): + while mock_aenter_client.await_count < 2: + await asyncio.sleep(0.01) + assert params.diagnostics.as_dict().get("reconnect_restart") == 1 + assert "reconnect_wait" not in params.diagnostics.as_dict() + finally: + await session.close() + + +async def test_disconnect_waits_for_backoff( + mock_aenter_client: AsyncMock, + message_iterator: FakeAsyncIterator, + mqtt_client_lite: AsyncMock, +) -> None: + """Test that a connection lost without a forced restart still waits before reconnecting.""" + params = copy.deepcopy(FAKE_PARAMS) + params.diagnostics = Diagnostics() + message_iterator.loop = False + with patch("roborock.mqtt.roborock_session.MIN_BACKOFF_INTERVAL", datetime.timedelta(hours=1)): + session = await create_mqtt_session(params) + await session.subscribe("topic-1", Subscriber().append) + + try: + await asyncio.sleep(0.1) + assert mock_aenter_client.await_count == 1 + assert params.diagnostics.as_dict().get("reconnect_wait", 0) >= 1 + assert "reconnect_restart" not in params.diagnostics.as_dict() + finally: + await session.close() + + +async def test_reconnect_resets_health_timeouts( + mock_aenter_client: AsyncMock, + message_iterator: FakeAsyncIterator, + mqtt_client_lite: AsyncMock, +) -> None: + """Test that timeouts counted before a reconnect don't restart the new connection.""" + params = copy.deepcopy(FAKE_PARAMS) + params.diagnostics = Diagnostics() + session = await create_mqtt_session(params) + await session.subscribe("topic-1", Subscriber().append) + health_manager = session.health_manager + + try: + await health_manager.on_timeout("device-1") + await health_manager.on_timeout("device-1") + + # The connection drops and the session reconnects on its own. + message_iterator.loop = False + async with asyncio.timeout(1): + while mock_aenter_client.await_count < 2: + await asyncio.sleep(0.01) + message_iterator.loop = True + + await health_manager.on_timeout("device-1") + assert "restart" not in params.diagnostics.as_dict() + finally: + await session.close() + + async def test_session_defers_reconnect_when_idle( mock_aenter_client: AsyncMock, message_iterator: FakeAsyncIterator, From ef781823c386fe1ad888646601a5dfaf1a30cd20 Mon Sep 17 00:00:00 2001 From: infy1995 Date: Fri, 13 Nov 2026 17:16:10 +0100 Subject: [PATCH 3/4] feat: add MQTT health to device diagnostics Add an mqtt_health section to the diagnostics of V1 and A01 devices: when the device started failing, its last success and timeout, request counts and its last 10 health events, newest first. Session restarts are recorded on every device with their cause. B01 devices don't report their requests to the health manager yet, so they get no section. --- roborock/devices/device.py | 7 +++++++ roborock/devices/device_manager.py | 7 ++++++- tests/devices/__snapshots__/test_device_manager.ambr | 9 +++++++++ tests/devices/test_device_manager.py | 3 +++ 4 files changed, 25 insertions(+), 1 deletion(-) diff --git a/roborock/devices/device.py b/roborock/devices/device.py index 44c275f6c..56d75fa50 100644 --- a/roborock/devices/device.py +++ b/roborock/devices/device.py @@ -15,6 +15,7 @@ from roborock.data import HomeDataDevice, HomeDataProduct from roborock.diagnostics import redact_device_data from roborock.exceptions import RoborockException +from roborock.mqtt.health_manager import DeviceHealthManager from roborock.roborock_message import RoborockMessage from roborock.util import RoborockLoggerAdapter @@ -59,12 +60,15 @@ def __init__( product: HomeDataProduct, channel: Channel, trait: Trait, + mqtt_health: DeviceHealthManager | None = None, ) -> None: """Initialize the RoborockDevice. The device takes ownership of the channel for communication with the device. Use `connect()` to establish the connection, which will set up the appropriate protocol channel. Use `close()` to clean up all connections. + + The optional `mqtt_health` adds the device's MQTT request history to its diagnostics. """ TraitsMixin.__init__(self, trait) self._duid = device_info.duid @@ -73,6 +77,7 @@ def __init__( self._device_info = device_info self._product = product self._channel = channel + self._mqtt_health = mqtt_health self._connect_task: asyncio.Task[None] | None = None self._unsub: Callable[[], None] | None = None self._ready_callbacks = CallbackList["RoborockDevice"]() @@ -257,6 +262,8 @@ def diagnostic_data(self) -> dict[str, Any]: extra["traits"] = self.v1_properties.as_dict() elif self.b01_q10_properties: extra["traits"] = self.b01_q10_properties.as_dict() + if self._mqtt_health: + extra["mqtt_health"] = self._mqtt_health.diagnostic_data() return redact_device_data( { "device": self.device_info.as_dict(), diff --git a/roborock/devices/device_manager.py b/roborock/devices/device_manager.py index f7d3a150a..f83c7b423 100644 --- a/roborock/devices/device_manager.py +++ b/roborock/devices/device_manager.py @@ -21,6 +21,7 @@ from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import B01Q10MapParserConfig from roborock.map.map_parser import MapParserConfig +from roborock.mqtt.health_manager import DeviceHealthManager from roborock.mqtt.roborock_session import create_lazy_mqtt_session from roborock.mqtt.session import MqttSession, SessionUnauthorizedHook from roborock.protocol import create_mqtt_params @@ -290,7 +291,11 @@ def device_creator(home_data: HomeData, device: HomeDataDevice, product: HomeDat f"Device {device.name} has unsupported version {device.pv} {product.model}" ) - dev = RoborockDevice(device, product, channel, trait) + # B01 channels don't report their MQTT requests to the health manager yet. + mqtt_health = ( + DeviceHealthManager(mqtt_session.health_manager, device.duid) if device.pv != DeviceVersion.B01 else None + ) + dev = RoborockDevice(device, product, channel, trait, mqtt_health) if ready_callback: dev.add_ready_callback(ready_callback) return dev diff --git a/tests/devices/__snapshots__/test_device_manager.ambr b/tests/devices/__snapshots__/test_device_manager.ambr index dadae5983..95963fe78 100644 --- a/tests/devices/__snapshots__/test_device_manager.ambr +++ b/tests/devices/__snapshots__/test_device_manager.ambr @@ -35,6 +35,15 @@ 'timeZoneId': 'America/Los_Angeles', 'tuyaMigrated': False, }), + 'mqtt_health': dict({ + 'events': list([ + ]), + 'failing_since': None, + 'last_success': None, + 'last_timeout': None, + 'successes': 0, + 'timeouts': 0, + }), 'product': dict({ 'capability': 0, 'category': 'robot.vacuum.cleaner', diff --git a/tests/devices/test_device_manager.py b/tests/devices/test_device_manager.py index 734c20657..0aaf87303 100644 --- a/tests/devices/test_device_manager.py +++ b/tests/devices/test_device_manager.py @@ -130,6 +130,9 @@ async def test_with_q10_device(cloud: FakeRoborockCloud, patch_device_manager: N assert q10_device.b01_q10_properties.status.status == YXDeviceState.CLEANING assert q10_sim.status[121] == 5 + # B01 requests are not tracked by the MQTT health manager, so no empty history is reported. + assert "mqtt_health" not in q10_device.diagnostic_data() + await device_manager.close() From bab8eb2fc6a3b525c89201c3e78adc663e333c28 Mon Sep 17 00:00:00 2001 From: infy1995 Date: Fri, 13 Nov 2026 17:16:10 +0100 Subject: [PATCH 4/4] docs: describe MQTT health management in the device architecture The health management section described local connection recovery as the HealthManager and stated that A01/B01 devices don't need one. Split it into local connection recovery and MQTT health management, covering the restart conditions and pacing and the mqtt_health diagnostics. --- docs/DEVICES.md | 48 +++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 41 insertions(+), 7 deletions(-) diff --git a/docs/DEVICES.md b/docs/DEVICES.md index cc7591e6b..6e148f25d 100644 --- a/docs/DEVICES.md +++ b/docs/DEVICES.md @@ -496,18 +496,52 @@ For each V1 command: | **Channel Class** | `V1Channel` | `MqttChannel` directly | | **RPC Abstraction** | `RpcChannel` with strategies | Helper functions | | **Strategy Pattern** | ✅ Multi-strategy (Local → MQTT) | ❌ Direct MQTT only | -| **Health Manager** | ✅ Tracks local/MQTT health | ❌ Not needed | +| **MQTT Health Reporting** | ✅ MQTT requests | ✅ A01 requests; ❌ B01 not yet | | **Code Location** | `v1_channel.py` | `a01_channel.py`, `b01_q7_channel.py` | -#### Health Management (V1 Only) +#### Local Connection Recovery (V1 Only) -Each V1 RPC strategy can have a `HealthManager` that tracks success/failure: +`V1Channel` keeps the local connection alive in the background: -- **Exponential Backoff**: After failures, wait before retrying -- **Automatic Recovery**: Periodically attempt to restore failed connections -- **Network Info Refresh**: Refresh local IP addresses after extended periods +- **Exponential Backoff**: After a failed local connection, wait before retrying (1 to 10 minutes) +- **Automatic Recovery**: Check the local connection periodically and reconnect when it drops +- **Network Info Refresh**: Refresh the device's local IP address after 12 hours -A01/B01 devices don't need health management since they only use MQTT (no fallback). +#### MQTT Health Management + +The broker can stop delivering one device's messages while the MQTT connection itself stays up and +other devices on the shared session keep working. The client keepalive can't detect this, so the +outcome of MQTT requests is tracked instead. Subscribing to the device's topic again on the same +connection doesn't bring its messages back; a new connection does. + +The session's `HealthManager` receives the result of every MQTT request through a per-device +`DeviceHealthManager`, and restarts the whole session (a new connection with all subscriptions +re-established) when: + +- **One device fails**: 3 of its requests time out within 30 minutes. A single response in between + doesn't reset the count, because a failing connection may still deliver some of them. The device + counts as recovered after 3 responses in a row, or at its first response once its timeouts are + older than 30 minutes. +- **The session fails**: 3 requests in a row time out across more than one device. + +Only timeouts on the current connection count: every new connection, whether forced or after the +session reconnected on its own, starts counting from zero. + +Restarts are paced so that a device that is really offline doesn't keep interrupting the others: + +- At most one restart every 30 minutes. +- While the same device keeps failing, the interval between the restarts it triggers doubles + (30 minutes, 1, 2, 4 hours, then every 6 hours). It is reset by the device's next response. +- A forced restart reconnects right away; other disconnects keep the exponential backoff. + +V1 devices answer most requests over the local connection, so their MQTT topic may carry no traffic +for hours. A broken topic is then noticed when the map (only available over MQTT) is requested +again, and the session is restarted after the first few map requests time out. + +Each device's diagnostics include an `mqtt_health` section: when it started failing, its last +success and timeout, request counts, and the last 10 health events (newest first): `timeouts_started`, +`session_restart` (with `cause`: `this_device`, `other_device` or `several_devices`) and `recovered`. +B01 devices don't report their requests yet, so they have no `mqtt_health` section. ### Protocol Versions