From dfdc01e353535a7403b6916c2631dfd2d9b9624d Mon Sep 17 00:00:00 2001 From: Leandro Pineda Date: Tue, 9 Jun 2026 13:25:23 -0300 Subject: [PATCH 1/4] Prevent MQTT callback errors from silently killing the network loop thread paho dispatches user callbacks (on_message, on_connect, on_disconnect, ...) on the network loop thread and, with `suppress_exceptions=False` (the default), re-raises any exception that escapes them. That re-raise unwinds `loop_forever` and the loop thread exits permanently -- it is never restarted, yet `is_connected()` can keep reporting True, so the session silently wedges (QoS-0 publishes are dropped) until the process restarts. - `_on_message` no longer re-raises: log with traceback and swallow, so a message-handler error can't take down the loop thread. - Set `client.suppress_exceptions = True` as a catch-all covering every callback (on_connect's subscribe/_resend_modules, on_disconnect, etc.). - Raise the default keepalive from 10s to 60s; 10s trips spurious "keep alive timeout" disconnects whenever the loop thread is briefly busy. Still overridable via the `keepalive_secs` kwarg. Tests: 2 new cases in test_robot_session_callbacks.py. Co-Authored-By: Claude Opus 4.8 (1M context) --- inorbit_edge/robot.py | 24 +++++++++---- .../tests/test_robot_session_callbacks.py | 36 +++++++++++++++++++ 2 files changed, 54 insertions(+), 6 deletions(-) diff --git a/inorbit_edge/robot.py b/inorbit_edge/robot.py index 7353da3..c9acff9 100644 --- a/inorbit_edge/robot.py +++ b/inorbit_edge/robot.py @@ -314,7 +314,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 +355,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. @@ -409,6 +411,11 @@ def __init__(self, robot_id, robot_name, api_key=None, **kwargs) -> None: self.client.on_message = self._on_message self.client.on_disconnect = self._on_disconnect + # Belt-and-suspenders: make paho swallow (not re-raise) any exception + # escaping a user callback, so a callback bug can never kill the network + # loop thread and silently wedge the session. + self.client.suppress_exceptions = True + # Functions to handle incoming MQTT messages. # They are mapped by MQTT subtopic e.g. # 'ros/loc/set_pose': set_pose_message_handler @@ -638,9 +645,14 @@ 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 + # Never re-raise: this callback runs on paho's network loop thread, + # and an exception escaping it kills that thread permanently (it is + # never restarted, while is_connected() can keep reporting True), + # silently wedging the session. Log with traceback and swallow. + self.logger.exception( + f"Unexpected error while 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..b087657 100644 --- a/inorbit_edge/tests/test_robot_session_callbacks.py +++ b/inorbit_edge/tests/test_robot_session_callbacks.py @@ -207,6 +207,42 @@ 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 + + @pytest.mark.parametrize( "test_input,expected", [ From 4c4a1229572d04a4f11de5cda86ff9aeb223eae8 Mon Sep 17 00:00:00 2001 From: Leandro Pineda Date: Tue, 9 Jun 2026 13:35:31 -0300 Subject: [PATCH 2/4] Guard MQTT callbacks and surface paho's internal logs Builds on suppress_exceptions to also keep the error visible. Without enable_logger, paho's own "Caught exception in ..." message (emitted when it swallows a callback exception) went nowhere, so suppress_exceptions alone kept the loop alive but discarded the error. - Add RobotSession._guard_callback(name) and wrap on_connect / on_message / on_disconnect at registration: an exception is logged with a traceback via the session logger (tagged with the callback name) and swallowed, so it never reaches paho's re-raise path on the network loop thread. - Call client.enable_logger(self.logger) so paho's internal logs are surfaced. - Keep suppress_exceptions = True as the catch-all net for unwrapped/future callbacks. Tests: 2 new cases (guard logs-and-swallows; enable_logger is wired). Co-Authored-By: Claude Opus 4.8 (1M context) --- inorbit_edge/robot.py | 58 ++++++++++++++++--- .../tests/test_robot_session_callbacks.py | 29 ++++++++++ 2 files changed, 80 insertions(+), 7 deletions(-) diff --git a/inorbit_edge/robot.py b/inorbit_edge/robot.py index c9acff9..d8f687f 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 @@ -406,14 +407,24 @@ 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 user callback, so a callback bug can never kill the network - # loop thread and silently wedge the session. + # Belt-and-suspenders for any callback path not wrapped above (and future + # ones): make paho swallow (not re-raise) any exception escaping a + # callback, so a callback bug can never kill the network loop thread. self.client.suppress_exceptions = True # Functions to handle incoming MQTT messages. @@ -625,6 +636,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. diff --git a/inorbit_edge/tests/test_robot_session_callbacks.py b/inorbit_edge/tests/test_robot_session_callbacks.py index b087657..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 @@ -243,6 +244,34 @@ def test_keepalive_default_is_60_seconds( 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", [ From ec347c6c0a3ab44332c3943a3d6e64818c1ce43f Mon Sep 17 00:00:00 2001 From: Leandro Pineda Date: Tue, 9 Jun 2026 14:23:01 -0300 Subject: [PATCH 3/4] Clarify why suppress_exceptions covers more than the wrapped callbacks Note in the comment that the flag is the only thing covering callback paths the per-callback guards do not: QoS>0 on_publish, consumer-set callbacks on the public client, and future callbacks added without a guard. Co-Authored-By: Claude Opus 4.8 (1M context) --- inorbit_edge/robot.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/inorbit_edge/robot.py b/inorbit_edge/robot.py index d8f687f..450803f 100644 --- a/inorbit_edge/robot.py +++ b/inorbit_edge/robot.py @@ -422,9 +422,12 @@ def __init__(self, robot_id, robot_name, api_key=None, **kwargs) -> None: self._on_disconnect ) - # Belt-and-suspenders for any callback path not wrapped above (and future - # ones): make paho swallow (not re-raise) any exception escaping a - # callback, so a callback bug can never kill the network loop thread. + # 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. From 4c07e8209ca5206d636daf852ad8bffd0d26b579 Mon Sep 17 00:00:00 2001 From: Leandro Pineda Date: Tue, 9 Jun 2026 16:33:29 -0300 Subject: [PATCH 4/4] Correct stale comment in _on_message exception handler The comment claimed an exception escaping _on_message kills paho's loop thread. That was true before _guard_callback wrapped the callbacks; now the wrapper (and suppress_exceptions) already prevent that. The clause remains for message-scoped logging (with topic) and to keep _on_message safe if called unwrapped -- not for loop-thread safety. Comment updated to match. Co-Authored-By: Claude Opus 4.8 (1M context) --- inorbit_edge/robot.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/inorbit_edge/robot.py b/inorbit_edge/robot.py index 450803f..097b0df 100644 --- a/inorbit_edge/robot.py +++ b/inorbit_edge/robot.py @@ -692,12 +692,14 @@ def _on_message(self, client, userdata, msg): f"Failed to decode message, ignoring. Payload: '{msg.payload}'. {ex}" ) except Exception: - # Never re-raise: this callback runs on paho's network loop thread, - # and an exception escaping it kills that thread permanently (it is - # never restarted, while is_connected() can keep reporting True), - # silently wedging the session. Log with traceback and swallow. + # 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"Unexpected error while processing message on topic " + f"Error processing message on topic " f"'{getattr(msg, 'topic', None)}', ignoring." )