diff --git a/inorbit_edge/robot.py b/inorbit_edge/robot.py index 7353da3..097b0df 100644 --- a/inorbit_edge/robot.py +++ b/inorbit_edge/robot.py @@ -1,5 +1,6 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- +import functools import io import json import logging @@ -314,7 +315,7 @@ def __init__(self, robot_id, robot_name, api_key=None, **kwargs) -> None: use_ssl (bool): Configures MQTT client to use SSL. Defaults: True. rest_api_endpoint (str): The URL of the InOrbit REST API. Defaults: INORBIT_DEFAULT_API_URL. - keepalive_secs (int): Keepalive for MQTT connection (seconds). Default: 10. + keepalive_secs (int): Keepalive for MQTT connection (seconds). Default: 60. estimate_distance_linear (bool): Whether to publish an estimate value for linear_distance based on poses when the value is not provided on a publish_odometry() call. Default: True @@ -355,8 +356,10 @@ def __init__(self, robot_id, robot_name, api_key=None, **kwargs) -> None: # transport if the environment variable HTTP_PROXY is set. self.use_websockets = kwargs.get("use_websockets", False) - # Keepalive for MQTT connection (seconds) - self.keepalive_secs = kwargs.get("keepalive_secs", 10) + # Keepalive for MQTT connection (seconds). 60s is a non-aggressive + # default; a very short keepalive trips spurious "keep alive timeout" + # disconnects whenever the network loop thread is briefly busy. + self.keepalive_secs = kwargs.get("keepalive_secs", 60) # Read optional proxy configuration from environment variables # We use ``self.http_proxy`` to indicate if proxy configuration should be used. @@ -404,10 +407,28 @@ def __init__(self, robot_id, robot_name, api_key=None, **kwargs) -> None: proxy_type=socks.HTTP, proxy_addr=proxy_hostname, proxy_port=proxy_port ) - # Register MQTT client callbacks - self.client.on_connect = self._on_connect - self.client.on_message = self._on_message - self.client.on_disconnect = self._on_disconnect + # Surface paho's internal logs (including the messages it emits when it + # swallows a callback exception under suppress_exceptions) through this + # session's logger, so they are not lost. + self.client.enable_logger(self.logger) + + # Register MQTT client callbacks, each wrapped so an exception is logged + # (with traceback) and swallowed rather than propagating onto the paho + # network loop thread -- where, by default, it would be re-raised and + # kill the thread, silently wedging the session. See _guard_callback. + self.client.on_connect = self._guard_callback("on_connect")(self._on_connect) + self.client.on_message = self._guard_callback("on_message")(self._on_message) + self.client.on_disconnect = self._guard_callback("on_disconnect")( + self._on_disconnect + ) + + # Belt-and-suspenders: make paho swallow (not re-raise) any exception + # escaping a callback, so a callback bug can never kill the network loop + # thread. This is the only thing covering callback paths the wrappers + # above do NOT -- e.g. QoS>0 publish acks invoking on_publish, callbacks + # a consumer sets directly on this client (which is a public attribute), + # or any callback added in the future without a guard. + self.client.suppress_exceptions = True # Functions to handle incoming MQTT messages. # They are mapped by MQTT subtopic e.g. @@ -618,6 +639,39 @@ def _on_connect(self, client, userdata, flags, reason_code, properties): # ask server to resend modules, so our state is consistent with the server side self._resend_modules() + def _guard_callback(self, name): + """Wrap a paho MQTT callback so an exception is logged (with traceback) + and swallowed instead of propagating onto paho's network loop thread. + + paho invokes callbacks on the loop thread and, with + ``suppress_exceptions`` False (its default), re-raises anything that + escapes them -- which unwinds ``loop_forever`` and kills the thread + permanently (it is never restarted, while ``is_connected()`` can keep + reporting True), silently wedging the session. This logs through the + session's own logger with the callback name and a traceback, so the + error is visible rather than merely suppressed. + + Args: + name (str): Callback name, used in the log message. + + Returns: + A decorator that wraps the callback. + """ + + def decorate(fn): + @functools.wraps(fn) + def wrapper(*args, **kwargs): + try: + return fn(*args, **kwargs) + except Exception: + self.logger.exception( + "Unhandled error in MQTT callback '%s', ignoring.", name + ) + + return wrapper + + return decorate + def _on_message(self, client, userdata, msg): """MQTT client message callback. @@ -638,9 +692,16 @@ def _on_message(self, client, userdata, msg): f"Failed to decode message, ignoring. Payload: '{msg.payload}'. {ex}" ) except Exception: - # Re-raise any other error - self.logger.error("Unexpected error while processing message.") - raise + # A handler (or echo) raised while processing this message. Log it + # with the topic for context and keep going. Re-raising is not + # needed for loop-thread safety here -- _guard_callback (and, behind + # it, suppress_exceptions) already stops a callback exception from + # unwinding paho's loop thread. This clause just adds a + # message-scoped log and keeps _on_message safe if called unwrapped. + self.logger.exception( + f"Error processing message on topic " + f"'{getattr(msg, 'topic', None)}', ignoring." + ) def _on_disconnect( self, client, userdata, disconnect_flags, reason_code, properties diff --git a/inorbit_edge/tests/test_robot_session_callbacks.py b/inorbit_edge/tests/test_robot_session_callbacks.py index f771926..f0838b3 100644 --- a/inorbit_edge/tests/test_robot_session_callbacks.py +++ b/inorbit_edge/tests/test_robot_session_callbacks.py @@ -1,6 +1,7 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- +import logging import os import time from unittest.mock import ANY, MagicMock @@ -207,6 +208,70 @@ def my_command_handler(*_): ) +def test_on_message_handler_error_does_not_propagate( + mock_mqtt_client, mock_inorbit_api, mock_sleep +): + """A message handler that raises must NOT propagate out of _on_message. + + _on_message runs on paho's network loop thread; an exception escaping it + kills that thread, which is never restarted, so the session silently wedges + (is_connected() can keep reporting True). The error must be logged and + swallowed instead of re-raised. + """ + robot_session = RobotSession( + robot_id="id_123", robot_name="name_123", api_key="apikey_123" + ) + + def boom(_payload): + raise RuntimeError("handler blew up") + + robot_session.message_handlers["my/sub"] = boom + + msg = MQTTMessage(topic=b"r/id_123/my/sub") + msg.payload = b"payload" + + # Must not raise. + robot_session._on_message(None, None, msg) + + +def test_keepalive_default_is_60_seconds( + mock_mqtt_client, mock_inorbit_api, mock_sleep +): + """Default MQTT keepalive should be a non-aggressive 60s, not 10s.""" + robot_session = RobotSession( + robot_id="id_123", robot_name="name_123", api_key="apikey_123" + ) + assert robot_session.keepalive_secs == 60 + + +def test_callback_guard_logs_and_swallows( + mock_mqtt_client, mock_inorbit_api, mock_sleep, caplog +): + """_guard_callback must log (with traceback) and swallow any exception so + a callback error never propagates onto paho's network loop thread.""" + robot_session = RobotSession( + robot_id="id_123", robot_name="name_123", api_key="apikey_123" + ) + + @robot_session._guard_callback("on_connect") + def boom(*_args, **_kwargs): + raise RuntimeError("callback blew up") + + caplog.set_level(logging.ERROR) + boom("client", "userdata") # must not raise + + assert any("on_connect" in r.getMessage() for r in caplog.records) + + +def test_paho_internal_logging_is_wired(mock_mqtt_client, mock_inorbit_api, mock_sleep): + """enable_logger must be called so paho's own callback-exception logs + (emitted when suppress_exceptions swallows) are surfaced, not lost.""" + robot_session = RobotSession( + robot_id="id_123", robot_name="name_123", api_key="apikey_123" + ) + robot_session.client.enable_logger.assert_called_once() + + @pytest.mark.parametrize( "test_input,expected", [