fix(mqtt): reconnect automatically when broker connection drops

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "openf1"
-version = "1.9.12"
+version = "1.9.13"
 authors = [
   { name="Bruno Godefroy" }
 ]

src/openf1/util/mqtt.py

@@ -1,11 +1,12 @@
 import os
 import ssl
 
-from aiomqtt import Client, MqttError
+from aiomqtt import Client
 from loguru import logger
 
 _url = os.getenv("OPENF1_MQTT_URL")
 _port_str = os.getenv("OPENF1_MQTT_PORT")
+_port = int(_port_str)
 _username = os.getenv("OPENF1_MQTT_USERNAME")
 _password = os.getenv("OPENF1_MQTT_PASSWORD")
 _disable_tls = os.getenv("OPENF1_MQTT_NO_TLS")
@@ -14,36 +15,41 @@
     _tls_context = None
 else:
     _tls_context = ssl.create_default_context()
+
 _client: Client | None = None
 
 
-async def initialize_mqtt():
-    """
-    Initializes and connects the global MQTT client.
-    Should be called once when the application starts.
-    """
+async def _connect() -> None:
+    """Tear down any existing client and open a fresh one.
+    Never raises; leaves _client as None if connecting fails."""
+    global _client
+    old, _client = _client, None
+    if old is not None:
+        try:
+            await old.__aexit__(None, None, None)
+        except Exception:
+            pass
+    client = Client(
+        hostname=_url,
+        port=_port,
+        username=_username,
+        password=_password,
+        tls_context=_tls_context,
+    )
+    try:
+        await client.__aenter__()
+        _client = client
+        logger.info("Connected to MQTT broker")
+    except Exception as e:
+        logger.error(f"Failed to connect to MQTT broker: {e}")
+
+
+async def initialize_mqtt() -> None:
+    """Connect at startup. Publish will also reconnect on its own if needed."""
     if not _url:
         logger.info("MQTT credentials not found, MQTT is disabled")
         return
-
-    global _client
-    if _client is None:
-        logger.info("Initializing MQTT client...")
-        _client = Client(
-            hostname=_url,
-            port=int(_port_str),
-            username=_username,
-            password=_password,
-            tls_context=_tls_context,
-        )
-        try:
-            await _client.__aenter__()
-            logger.info("Successfully connected to MQTT broker")
-        except (MqttError, ValueError, OSError) as e:
-            logger.error(f"Failed to connect to MQTT broker: {e}")
-            _client = None
-    else:
-        logger.info("MQTT client is already initialized")
+    await _connect()
 
 
 async def publish_messages_to_mqtt(
@@ -52,6 +58,7 @@ async def publish_messages_to_mqtt(
     """
     Publishes multiple messages to an MQTT topic asynchronously.
     Assumes initialize_mqtt() has already been called.
+    Never raises; on failure it drops the batch and returns False.
 
     Args:
         topic: The MQTT topic to publish to
@@ -65,13 +72,16 @@ async def publish_messages_to_mqtt(
         logger.warning("No messages to publish")
         return True
 
+    if _client is None:
+        await _connect()
+    if _client is None:
+        return False
+
     try:
         for message in messages:
             await _client.publish(topic, payload=message, qos=qos)
         return True
-    except MqttError as e:
-        logger.exception(f"MQTT error while publishing to topic '{topic}': {e}")
-        return False
-    except Exception:
-        logger.exception("An unexpected error occurred during MQTT publish")
+    except Exception as e:
+        logger.warning(f"MQTT publish to '{topic}' failed, dropping batch: {e}")
+        await _connect()  # reopen for the next call
         return False