diff --git a/CHANGELOG.md b/CHANGELOG.md index 79bcf65..d09ab97 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -51,6 +51,7 @@ semantic versioning. ### Fixed +- The weather service's Blitzortung lightning feed backs off when the broker keeps dropping it: 5 seconds after a drop, doubling to 5 minutes, and back to 5 seconds after a connection that lasts a minute. It used to reconnect every second, logging two lines each time ([#362](https://github.com/agessaman/meshcore-bot/issues/362)). It only logs "Connected" once the broker accepts the session, says why it is reconnecting, and gives each connection its own client ID. Strikes inside `blitz_area_*` also reach the lightning handler now. They arrive on the MQTT library's own thread, where scheduling the handler failed and was logged only at debug level, so they were dropped. - The web viewer's Email & Notifications panel has an "Allow a local or private-network SMTP server" checkbox. `allow_local_smtp` is stored in the bot database, not `config.ini`, and no control set it, so a local relay such as Postfix could not be used for the nightly email or radio alerts ([#361](https://github.com/agessaman/meshcore-bot/issues/361)). - A repeater heard twice in a row no longer posts two "new repeater" messages to Discord and Telegram. The prefix-collision service decided to announce a key and only recorded that key after the posts returned, so a second NEW_CONTACT while the first post was still in flight sent again (and the same gap double-posted a prefix-collision alert onto the mesh, which the bridges delivered twice). It now records the key before it sends. - The greeter's warmup (`rollout_days`) runs once, the first time the greeter starts. A restart more than a day after the previous warmup ended used to open another one, and anyone who spoke during it was marked greeted and never welcomed. Ending a warmup, or it expiring, no longer starts a new one on the next boot. diff --git a/modules/service_plugins/weather_service.py b/modules/service_plugins/weather_service.py index 7283b96..7e54bd7 100644 --- a/modules/service_plugins/weather_service.py +++ b/modules/service_plugins/weather_service.py @@ -9,6 +9,7 @@ import contextlib # noqa: F401 importable from this module on dev import json import math import re +import secrets import threading import time import xml.dom.minidom @@ -58,6 +59,14 @@ FORECAST_RETRY_OFFSETS_SECONDS = (300, 900, 1800) FORECAST_RETRY_JOB_ID = "weather_daily_forecast_retry" TRANSIENT_FORECAST_HTTP_STATUSES = frozenset({429, 500, 502, 503, 504}) +# Blitzortung MQTT reconnects back off exponentially between these bounds. A +# session that stays up for BLITZ_STABLE_SESSION_SECONDS resets the delay, so a +# broker that drops every connection within seconds is retried less and less often. +BLITZ_RECONNECT_MIN_SECONDS = 5.0 +BLITZ_RECONNECT_MAX_SECONDS = 300.0 +BLITZ_STABLE_SESSION_SECONDS = 60.0 +BLITZ_CONNACK_TIMEOUT_SECONDS = 10.0 + @dataclass(frozen=True) class ForecastFetchResult: @@ -1264,14 +1273,20 @@ class WeatherService(BaseServicePlugin): self.logger.info(f"Connecting to Blitzortung MQTT broker: {broker_host}:{broker_port}") + delay = BLITZ_RECONNECT_MIN_SECONDS while self._running: + connected_at: Optional[float] = None try: - # Create paho-mqtt client - client_id = f"meshcore_weather_{int(time.time())}" + # The random suffix keeps two bots started in the same second + # from sharing an ID, which makes the broker drop the older one. + client_id = f"meshcore_weather_{int(time.time())}_{secrets.token_hex(3)}" client = mqtt.Client(client_id=client_id) self.mqtt_client = client + disconnect_rc: list[Any] = [] + loop = asyncio.get_running_loop() - # Set up message callback + # paho calls this on its network thread, so the strike is + # handed to the bot's event loop rather than scheduled here. def on_message(client, userdata, msg): try: # Decode message @@ -1287,8 +1302,7 @@ class WeatherService(BaseServicePlugin): if (self.blitz_area['min_lat'] <= lat <= self.blitz_area['max_lat'] and self.blitz_area['min_lon'] <= lon <= self.blitz_area['max_lon']): - # Schedule async processing - asyncio.create_task(self._handle_lightning_strike(blitz_data)) + asyncio.run_coroutine_threadsafe(self._handle_lightning_strike(blitz_data), loop) except json.JSONDecodeError: self.logger.debug("Invalid JSON in lightning MQTT message") @@ -1296,13 +1310,12 @@ class WeatherService(BaseServicePlugin): self.logger.debug(f"Error processing lightning MQTT message: {e}") client.on_message = on_message + client.on_disconnect = lambda _client, _userdata, rc: disconnect_rc.append(rc) # Connect and subscribe (non-blocking to avoid blocking event loop) - loop = asyncio.get_running_loop() try: await loop.run_in_executor(None, client.connect, broker_host, broker_port, 60) except Exception as connect_error: - # Connection failed, but don't block - will retry on next cycle self.logger.debug(f"Initial connect() call failed (non-blocking): {connect_error}") raise # Re-raise to trigger retry logic @@ -1315,26 +1328,47 @@ class WeatherService(BaseServicePlugin): client.loop_start() - self.logger.info(f"Connected to Blitzortung MQTT, subscribed to {topic}") + # connect() only opens the socket; the session is up once the + # broker's CONNACK arrives on the network thread. + deadline = time.monotonic() + BLITZ_CONNACK_TIMEOUT_SECONDS + while self._running and not client.is_connected() and time.monotonic() < deadline: + await asyncio.sleep(0.25) - # Keep connection alive - while self._running: - await asyncio.sleep(1) - if not client.is_connected(): - self.logger.warning("Blitzortung MQTT disconnected, reconnecting...") - break + if client.is_connected(): + connected_at = time.monotonic() + self.logger.info(f"Connected to Blitzortung MQTT, subscribed to {topic}") + while self._running and client.is_connected(): + await asyncio.sleep(1) client.loop_stop() client.disconnect() + if not self._running: + break + + if connected_at is None: + reason = "broker did not accept the connection" + else: + reason = "disconnected" + if time.monotonic() - connected_at >= BLITZ_STABLE_SESSION_SECONDS: + delay = BLITZ_RECONNECT_MIN_SECONDS + if disconnect_rc: + reason += f" (rc={disconnect_rc[-1]})" + self.logger.warning(f"Blitzortung MQTT {reason}, reconnecting in {delay:.0f}s") except asyncio.CancelledError: break except Exception as e: self.logger.error(f"Error in Blitzortung MQTT connection: {e}") - if self._running: - self.logger.info("Reconnecting to Blitzortung MQTT in 30 seconds...") - await asyncio.sleep(30) + if not self._running: + break + self.logger.info(f"Reconnecting to Blitzortung MQTT in {delay:.0f}s") + + try: + await asyncio.sleep(delay) + except asyncio.CancelledError: + break + delay = min(delay * 2, BLITZ_RECONNECT_MAX_SECONDS) async def _handle_lightning_strike(self, blitz_data: dict[str, Any]) -> None: """Handle a single lightning strike from MQTT. diff --git a/tests/unit/test_weather_blitz_reconnect.py b/tests/unit/test_weather_blitz_reconnect.py new file mode 100644 index 0000000..3db0c83 --- /dev/null +++ b/tests/unit/test_weather_blitz_reconnect.py @@ -0,0 +1,165 @@ +#!/usr/bin/env python3 +"""Unit tests for the Blitzortung MQTT reconnect loop and strike handoff (#362).""" + +import asyncio +import configparser +import json +import threading +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +from modules.service_plugins import weather_service as ws +from modules.service_plugins.weather_service import WeatherService + +pytestmark = pytest.mark.unit + +BOX = { + "blitz_area_min_lat": "47.0", + "blitz_area_min_lon": "-123.0", + "blitz_area_max_lat": "48.5", + "blitz_area_max_lon": "-121.5", +} + + +class FakeClient: + """A paho client whose session lasts ``session_checks`` is_connected() calls.""" + + instances: list["FakeClient"] = [] + session_checks = 2 + + def __init__(self, client_id): + self.client_id = client_id + self.on_message = None + self.on_disconnect = None + self._checks_left = 0 + FakeClient.instances.append(self) + + def connect(self, host, port, keepalive): + pass + + def subscribe(self, topic): + pass + + def loop_start(self): + self._checks_left = FakeClient.session_checks + + def is_connected(self): + if self._checks_left > 0: + self._checks_left -= 1 + return True + return False + + def loop_stop(self): + pass + + def disconnect(self): + pass + + +@pytest.fixture +def service(monkeypatch, mock_logger): + FakeClient.instances = [] + FakeClient.session_checks = 2 + monkeypatch.setattr(ws, "mqtt", SimpleNamespace(Client=FakeClient), raising=False) + monkeypatch.setattr(ws, "MQTT_AVAILABLE", True) + + config = configparser.ConfigParser() + config.add_section("Weather") + config.add_section("Weather_Service") + config.set("Weather_Service", "my_position_lat", "47.6062") + config.set("Weather_Service", "my_position_lon", "-122.3321") + for key, val in BOX.items(): + config.set("Weather_Service", key, val) + + bot = Mock() + bot.logger = mock_logger + bot.config = config + bot.db_manager = Mock() + bot.command_manager = Mock() + bot.command_manager.send_channel_message = AsyncMock() + svc = WeatherService(bot) + svc._running = True + return svc + + +def record_backoff(monkeypatch, service, attempts): + """Make sleeps instant; record the backoff ones and stop after ``attempts``.""" + real_sleep = asyncio.sleep + delays: list[float] = [] + + async def fake_sleep(seconds): + if seconds >= ws.BLITZ_RECONNECT_MIN_SECONDS: + delays.append(seconds) + if len(delays) >= attempts: + service._running = False + await real_sleep(0) + + monkeypatch.setattr(ws.asyncio, "sleep", fake_sleep) + return delays + + +class TestReconnectBackoff: + async def test_flapping_broker_backs_off_to_the_cap(self, monkeypatch, service): + delays = record_backoff(monkeypatch, service, attempts=8) + await asyncio.wait_for(service._connect_blitzortung_mqtt(), 5) + assert delays == [5, 10, 20, 40, 80, 160, 300, 300] + + async def test_stable_session_resets_the_delay(self, monkeypatch, service): + monkeypatch.setattr(ws, "BLITZ_STABLE_SESSION_SECONDS", 0.0) + delays = record_backoff(monkeypatch, service, attempts=4) + await asyncio.wait_for(service._connect_blitzortung_mqtt(), 5) + assert delays == [5, 5, 5, 5] + + async def test_unaccepted_connection_is_not_logged_as_connected(self, monkeypatch, service, mock_logger): + FakeClient.session_checks = 0 + monkeypatch.setattr(ws, "BLITZ_CONNACK_TIMEOUT_SECONDS", 0.0) + delays = record_backoff(monkeypatch, service, attempts=3) + await asyncio.wait_for(service._connect_blitzortung_mqtt(), 5) + assert delays == [5, 10, 20] + infos = [c.args[0] for c in mock_logger.info.call_args_list] + assert not any("Connected to Blitzortung" in m for m in infos) + warnings = [c.args[0] for c in mock_logger.warning.call_args_list] + assert "Blitzortung MQTT broker did not accept the connection, reconnecting in 5s" in warnings + + async def test_connect_error_backs_off(self, monkeypatch, service): + monkeypatch.setattr(FakeClient, "connect", Mock(side_effect=OSError("refused"))) + delays = record_backoff(monkeypatch, service, attempts=3) + await asyncio.wait_for(service._connect_blitzortung_mqtt(), 5) + assert delays == [5, 10, 20] + + async def test_client_ids_are_unique_within_a_second(self, monkeypatch, service): + record_backoff(monkeypatch, service, attempts=5) + await asyncio.wait_for(service._connect_blitzortung_mqtt(), 5) + ids = [c.client_id for c in FakeClient.instances] + assert len(ids) == 5 + assert len(set(ids)) == 5 + + +class TestStrikeHandoff: + async def test_strike_from_network_thread_reaches_the_handler(self, monkeypatch, service): + handled = asyncio.Event() + received = [] + + async def handle(data): + received.append(data) + handled.set() + + monkeypatch.setattr(service, "_handle_lightning_strike", handle) + FakeClient.session_checks = 10_000 + task = asyncio.create_task(service._connect_blitzortung_mqtt()) + while not FakeClient.instances or FakeClient.instances[0].on_message is None: + await asyncio.sleep(0) + + on_message = FakeClient.instances[0].on_message + strike = {"lat": 47.6, "lon": -122.3, "time": 1} + msg = SimpleNamespace(payload=json.dumps(strike).encode()) + thread = threading.Thread(target=on_message, args=(None, None, msg)) + thread.start() + thread.join() + + await asyncio.wait_for(handled.wait(), 2) + assert received == [strike] + service._running = False + await asyncio.wait_for(task, 5)