fix: Keep MQTT topic subscriptions alive with an idle timeout #632
Conversation
There was a problem hiding this comment.
Pull request overview
This PR implements an idle timeout mechanism for MQTT topic subscriptions to keep connections alive across multiple RPCs. When the last callback unsubscribes from a topic, instead of immediately unsubscribing from the MQTT broker, the session waits for a configurable idle timeout period (default 60 seconds). If a new subscription occurs during this period, the timer is cancelled and the existing subscription is reused, avoiding unnecessary reconnection overhead.
Key changes:
- Added configurable
topic_idle_timeoutparameter toRoborockMqttSessionwith default of 60 seconds - Modified
subscribe()to return a delayed unsubscribe function that schedules idle timeout instead of immediate unsubscription - Updated
close()to cancel all pending idle timers
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 5 comments.
| File | Description |
|---|---|
| roborock/mqtt/roborock_session.py | Implements idle timeout mechanism with timer management, modified subscribe/unsubscribe logic, and updated keepalive constants |
| tests/mqtt/test_roborock_session.py | Adds mock MQTT client fixture and three new tests validating idle timeout behavior (resubscribe cancellation, timeout expiration, multiple callbacks) |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| await asyncio.sleep(self._topic_idle_timeout.total_seconds()) | ||
| # Only unsubscribe if there are no callbacks left for this topic | ||
| if not self._listeners.get_callbacks(topic): | ||
| async with self._client_lock: | ||
| if self._client: | ||
| _LOGGER.debug("Idle timeout expired, unsubscribing from topic %s", topic) | ||
| try: | ||
| await self._client.unsubscribe(topic) | ||
| except MqttError as err: | ||
| _LOGGER.warning("Error unsubscribing from topic %s: %s", topic, err) |
There was a problem hiding this comment.
A race condition exists in the idle timer cleanup logic. When the timer expires and checks if not self._listeners.get_callbacks(topic), another task could add a new callback between this check and the actual unsubscribe call. This could lead to unsubscribing from a topic that has active callbacks.
To fix this, the check and the unsubscribe operation should be atomic within the lock. Consider moving the get_callbacks check inside the async with self._client_lock: block before unsubscribing.
Lash-L
left a comment
Lash-L
left a comment
There was a problem hiding this comment.
Looks good I think! Did not look at copilots comments
2 participants
Keep topic connections open across multiple RPCs. We currently use the "subscribe" and "publish" mechanism for sending RPCs, and so keeping that interface in use is simpler than adding another RPC Queue (after consultation with Clade). This keeps the responsibility to this lower session layer and will disconnect form a topic after an idle timeout of inactivity.
Issue #619