diff --git a/drb-edge-node/app/internal/health.py b/drb-edge-node/app/internal/health.py index 1268a67..66ea358 100644 --- a/drb-edge-node/app/internal/health.py +++ b/drb-edge-node/app/internal/health.py @@ -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).", "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: checks.append(_check( "mqtt", "Live link (MQTT)", ERROR, @@ -333,6 +349,7 @@ async def gather_facts() -> Dict[str, Any]: "c2_url": settings.c2_url, "has_api_key": bool(credentials.get_api_key()), "mqtt_connected": mqtt_manager.is_connected, + "mqtt_last_error": mqtt_manager.last_error, "mqtt_broker": settings.mqtt_broker, "mqtt_port": settings.mqtt_port, "sdr_count": sdr_count, diff --git a/drb-edge-node/app/internal/mqtt_manager.py b/drb-edge-node/app/internal/mqtt_manager.py index 09eb12a..52f29bd 100644 --- a/drb-edge-node/app/internal/mqtt_manager.py +++ b/drb-edge-node/app/internal/mqtt_manager.py @@ -1,6 +1,7 @@ import asyncio import json import shutil +import time from datetime import datetime, timezone from collections import deque from typing import Optional, Callable, Awaitable, Dict, Any @@ -60,6 +61,12 @@ class MQTTManager: # install.sh writing the file). Recovery compares against this. self._client_key: Optional[str] = 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_config_push: Optional[ConfigCallback] = None @@ -146,12 +153,21 @@ class MQTTManager: client.reconnect_delay_set(min_delay=2, max_delay=60) client.on_connect = self._on_connect client.on_disconnect = self._on_disconnect + client.on_connect_fail = self._on_connect_fail client.on_message = self._on_message 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): if reason_code == 0: self._connected = True + self._last_error = None client.subscribe(self._t_commands, qos=1) client.subscribe(self._t_config, 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) else: 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: # paho thread -> event loop. A refused login never fixes itself # 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): self._connected = False 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): try: @@ -273,6 +293,7 @@ class MQTTManager: return except Exception as e: 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) delay = min(delay * 2, 60) @@ -408,5 +429,10 @@ class MQTTManager: def is_connected(self) -> bool: 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() diff --git a/drb-edge-node/tests/test_local_dashboard.py b/drb-edge-node/tests/test_local_dashboard.py index 954fb23..e0ffe69 100644 --- a/drb-edge-node/tests/test_local_dashboard.py +++ b/drb-edge-node/tests/test_local_dashboard.py @@ -100,6 +100,17 @@ def test_key_on_disk_but_mqtt_down_says_restart(): 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(): c = _by_id(health.build_checks(_facts(c2_url=None, has_api_key=False, mqtt_connected=False))) assert c["c2"]["state"] == "off" diff --git a/drb-edge-node/tests/test_mqtt_manager.py b/drb-edge-node/tests/test_mqtt_manager.py index 96686d6..31414e6 100644 --- a/drb-edge-node/tests/test_mqtt_manager.py +++ b/drb-edge-node/tests/test_mqtt_manager.py @@ -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. from app.internal.mqtt_manager import disk_report 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