fix(weather): back off Blitzortung reconnects and deliver strikes

A broker that drops the session within a second had the service reconnecting every second indefinitely. Reconnects now back off from 5 seconds, doubling to a 5-minute cap, and reset after a session that stays up for a minute. "Connected" is logged only once the broker's CONNACK arrives rather than when the socket opens, the reconnect warning gives the reason and the paho disconnect code, and the client ID gets a random suffix so two bots started in the same second don't knock each other off.

on_message runs on paho's network thread, where asyncio.create_task has no running loop, so every in-area strike raised and was swallowed at debug level. It now hands the strike to the bot's loop with run_coroutine_threadsafe.

Fixes #362

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.

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.

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)