mqtt_manager: expose last connect error; health tells refused from unreachable (node-26#21)

mqtt_manager.last_error = {kind, detail, epoch}: 'refused' from a non-zero
CONNACK, 'unreachable' from the initial connect loop or paho's
on_connect_fail, 'disconnected' otherwise. Cleared on connect. The /api/health
MQTT check now gives a key fix for a refusal and a network fix for an
unreachable broker instead of guessing from whether a key is on disk.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Logan Cusano
2026-10-27 20:07:47 -04:00
co-authored by Claude Opus 5.5
parent a456b502db
commit 1b4fcefbf2
4 changed files with 79 additions and 0 deletions
+17
View File
@@ -154,6 +154,22 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]:
f"Not connected to {broker} — expected until the node is approved (the broker only admits approved nodes).", f"Not connected to {broker} — expected until the node is approved (the broker only admits approved nodes).",
"Fixes itself once the node is approved and has its API key (see above).", "Fixes itself once the node is approved and has its API key (see above).",
)) ))
elif (f.get("mqtt_last_error") or {}).get("kind") == "refused":
err = f["mqtt_last_error"]
checks.append(_check(
"mqtt", "Live link (MQTT)", ERROR,
f"{broker} refused this node's login ({err.get('detail')}).",
"The API key on file is not the one the broker expects — every re-approval issues a new one. "
"Re-run install.sh to fetch the current key, or ask a DRB admin to re-approve the node.",
))
elif (f.get("mqtt_last_error") or {}).get("kind") == "unreachable":
err = f["mqtt_last_error"]
checks.append(_check(
"mqtt", "Live link (MQTT)", ERROR,
f"Can't reach {broker} ({err.get('detail')}).",
f"Check this site's internet connection, DNS, and that outbound port {f['mqtt_port']} isn't "
"blocked by a firewall. The key is not the problem.",
))
else: else:
checks.append(_check( checks.append(_check(
"mqtt", "Live link (MQTT)", ERROR, "mqtt", "Live link (MQTT)", ERROR,
@@ -333,6 +349,7 @@ async def gather_facts() -> Dict[str, Any]:
"c2_url": settings.c2_url, "c2_url": settings.c2_url,
"has_api_key": bool(credentials.get_api_key()), "has_api_key": bool(credentials.get_api_key()),
"mqtt_connected": mqtt_manager.is_connected, "mqtt_connected": mqtt_manager.is_connected,
"mqtt_last_error": mqtt_manager.last_error,
"mqtt_broker": settings.mqtt_broker, "mqtt_broker": settings.mqtt_broker,
"mqtt_port": settings.mqtt_port, "mqtt_port": settings.mqtt_port,
"sdr_count": sdr_count, "sdr_count": sdr_count,
@@ -1,6 +1,7 @@
import asyncio import asyncio
import json import json
import shutil import shutil
import time
from datetime import datetime, timezone from datetime import datetime, timezone
from collections import deque from collections import deque
from typing import Optional, Callable, Awaitable, Dict, Any from typing import Optional, Callable, Awaitable, Dict, Any
@@ -60,6 +61,12 @@ class MQTTManager:
# install.sh writing the file). Recovery compares against this. # install.sh writing the file). Recovery compares against this.
self._client_key: Optional[str] = None self._client_key: Optional[str] = None
self._recovery_task: Optional[asyncio.Task] = None self._recovery_task: Optional[asyncio.Task] = None
# Why the link is down, for the dashboard (node-26#21): "refused" (the
# broker answered and said no), "unreachable" (TCP/TLS never got that
# far) or "disconnected" (an established session dropped). Cleared on
# every successful connect. Set from paho's thread as well as the event
# loop — a single attribute rebind, so no lock.
self._last_error: Optional[Dict[str, Any]] = None
self.on_command: Optional[CommandCallback] = None self.on_command: Optional[CommandCallback] = None
self.on_config_push: Optional[ConfigCallback] = None self.on_config_push: Optional[ConfigCallback] = None
@@ -146,12 +153,21 @@ class MQTTManager:
client.reconnect_delay_set(min_delay=2, max_delay=60) client.reconnect_delay_set(min_delay=2, max_delay=60)
client.on_connect = self._on_connect client.on_connect = self._on_connect
client.on_disconnect = self._on_disconnect client.on_disconnect = self._on_disconnect
client.on_connect_fail = self._on_connect_fail
client.on_message = self._on_message client.on_message = self._on_message
return client return client
def _set_error(self, kind: str, detail: str) -> None:
self._last_error = {"kind": kind, "detail": detail, "epoch": time.time()}
def _on_connect_fail(self, client, userdata):
# paho's own reconnect (after loop_start) could not open the socket.
self._set_error("unreachable", f"cannot reach {settings.mqtt_broker}:{settings.mqtt_port}")
def _on_connect(self, client, userdata, flags, reason_code, properties): def _on_connect(self, client, userdata, flags, reason_code, properties):
if reason_code == 0: if reason_code == 0:
self._connected = True self._connected = True
self._last_error = None
client.subscribe(self._t_commands, qos=1) client.subscribe(self._t_commands, qos=1)
client.subscribe(self._t_config, qos=1) client.subscribe(self._t_config, qos=1)
client.subscribe(self._t_discord_desired, qos=1) client.subscribe(self._t_discord_desired, qos=1)
@@ -164,6 +180,7 @@ class MQTTManager:
asyncio.run_coroutine_threadsafe(self._flush_offline_buffer(), self._loop) asyncio.run_coroutine_threadsafe(self._flush_offline_buffer(), self._loop)
else: else:
logger.error(f"MQTT connect refused: {reason_code}") logger.error(f"MQTT connect refused: {reason_code}")
self._set_error("refused", str(reason_code))
if _is_auth_failure(reason_code) and self._loop is not None: if _is_auth_failure(reason_code) and self._loop is not None:
# paho thread -> event loop. A refused login never fixes itself # paho thread -> event loop. A refused login never fixes itself
# by retrying the same password, so go get the right one. # by retrying the same password, so go get the right one.
@@ -172,6 +189,9 @@ class MQTTManager:
def _on_disconnect(self, client, userdata, disconnect_flags, reason_code, properties): def _on_disconnect(self, client, userdata, disconnect_flags, reason_code, properties):
self._connected = False self._connected = False
logger.warning(f"MQTT disconnected: {reason_code}") logger.warning(f"MQTT disconnected: {reason_code}")
# A refusal is followed by a disconnect; keep the more specific reason.
if self._last_error is None:
self._set_error("disconnected", str(reason_code))
def _on_message(self, client, userdata, msg): def _on_message(self, client, userdata, msg):
try: try:
@@ -273,6 +293,7 @@ class MQTTManager:
return return
except Exception as e: except Exception as e:
logger.warning(f"MQTT connect failed ({e}) — retrying in {delay}s") logger.warning(f"MQTT connect failed ({e}) — retrying in {delay}s")
self._set_error("unreachable", f"cannot reach {settings.mqtt_broker}:{settings.mqtt_port}: {e}")
await asyncio.sleep(delay) await asyncio.sleep(delay)
delay = min(delay * 2, 60) delay = min(delay * 2, 60)
@@ -408,5 +429,10 @@ class MQTTManager:
def is_connected(self) -> bool: def is_connected(self) -> bool:
return self._connected return self._connected
@property
def last_error(self) -> Optional[Dict[str, Any]]:
"""{kind, detail, epoch} for the most recent failure, None while connected."""
return None if self._connected else self._last_error
mqtt_manager = MQTTManager() mqtt_manager = MQTTManager()
@@ -100,6 +100,17 @@ def test_key_on_disk_but_mqtt_down_says_restart():
assert "restart the edge-node container" in c["mqtt"]["action"] assert "restart the edge-node container" in c["mqtt"]["action"]
def test_mqtt_refused_vs_unreachable_are_told_apart():
"""node-26#21: 'not authorised' and 'can't reach the broker' need different fixes."""
refused = _by_id(health.build_checks(_facts(
mqtt_connected=False, mqtt_last_error={"kind": "refused", "detail": "Not authorized", "epoch": 1.0})))
assert refused["mqtt"]["state"] == "error"
assert "refused" in refused["mqtt"]["summary"] and "install.sh" in refused["mqtt"]["action"]
unreachable = _by_id(health.build_checks(_facts(
mqtt_connected=False, mqtt_last_error={"kind": "unreachable", "detail": "timed out", "epoch": 1.0})))
assert "Can't reach" in unreachable["mqtt"]["summary"] and "firewall" in unreachable["mqtt"]["action"]
def test_standalone_node_is_not_an_error(): def test_standalone_node_is_not_an_error():
c = _by_id(health.build_checks(_facts(c2_url=None, has_api_key=False, mqtt_connected=False))) c = _by_id(health.build_checks(_facts(c2_url=None, has_api_key=False, mqtt_connected=False)))
assert c["c2"]["state"] == "off" assert c["c2"]["state"] == "off"
+25
View File
@@ -153,3 +153,28 @@ def test_disk_report_nulls_when_path_unreadable(tmp_path):
# node-26#41: nulls (not omitted keys) so C2 never keeps a stale reading. # node-26#41: nulls (not omitted keys) so C2 never keeps a stale reading.
from app.internal.mqtt_manager import disk_report from app.internal.mqtt_manager import disk_report
assert disk_report(str(tmp_path / "missing")) == {"disk_free_bytes": None, "disk_total_bytes": None} assert disk_report(str(tmp_path / "missing")) == {"disk_free_bytes": None, "disk_total_bytes": None}
def test_last_error_records_refusal_and_clears_on_connect(monkeypatch):
"""node-26#21: the refusal reason survives the disconnect that follows it."""
from unittest.mock import MagicMock
mqtt_manager._last_error = None
mqtt_manager._loop = None
mqtt_manager._on_connect(MagicMock(), None, None, 135, None)
mqtt_manager._on_disconnect(MagicMock(), None, None, 135, None)
assert mqtt_manager.last_error["kind"] == "refused"
monkeypatch.setattr(mqtt_manager, "_flush_offline_buffer", MagicMock())
monkeypatch.setattr(mqtt_manager, "_publish_checkin", MagicMock())
monkeypatch.setattr(mqtt_manager, "_maybe_request_key", MagicMock())
monkeypatch.setattr("asyncio.run_coroutine_threadsafe", lambda *a, **k: None)
mqtt_manager._on_connect(MagicMock(), None, None, 0, None)
assert mqtt_manager.last_error is None
mqtt_manager._connected = False
def test_last_error_unreachable_from_connect_fail():
mqtt_manager._last_error = None
mqtt_manager._connected = False
mqtt_manager._on_connect_fail(None, None)
assert mqtt_manager.last_error["kind"] == "unreachable"
mqtt_manager._last_error = None