fix: MQTT connection flapping + comprehensive bug fixes

CHANGELOG.rst

@@ -2,8 +2,70 @@
 Changelog
 =========
 
-Unreleased (8.x)
-================
+Unreleased
+==========
+
+Version 8.1.0 (2026-05-16)
+==========================
+
+Bug Fixes
+---------
+- **Fix MQTT connection flapping after reconnect**: When ``_active_reconnect()``
+  created a new ``MqttConnection``, the old connection was never closed. The old
+  SDK connection's built-in auto-reconnect would eventually succeed, creating two
+  active connections sharing the same client ID. Because AWS IoT allows only one
+  connection per client ID, the broker would kick one off, triggering
+  ``on_connection_interrupted`` and starting yet another reconnection — an
+  infinite connect/disconnect loop. Fixed by adding ``MqttConnection.close()``
+  (unconditional teardown regardless of ``_connected`` state) and calling it
+  before creating the replacement connection in both ``_active_reconnect()`` and
+  ``_deep_reconnect()``.
+
+- **Thread-safety race in ``ensure_device_info_cached``**: The ``future.done()``
+  check and ``future.set_result()`` were performed in the AWS SDK callback thread
+  without synchronisation, creating a race against the asyncio event loop thread.
+  Moved both operations inside a ``call_soon_threadsafe`` callback so they execute
+  atomically on the event loop thread.
+
+- **ZeroDivisionError when ``deep_reconnect_threshold`` is 0**: Config validation
+  now clamps ``deep_reconnect_threshold`` to a minimum of 1, preventing a
+  ``ZeroDivisionError`` in the exponential-backoff reconnection logic.
+
+- **Reconnect counter never incremented**: ``total_reconnect_attempts`` in
+  diagnostics was not incremented on connection drops, so it always reported 0
+  despite active reconnections. Counter is now incremented on each
+  ``on_connection_interrupted`` event.
+
+- **``shortest_session_seconds`` not JSON-serialisable**: The diagnostics
+  ``to_dict()`` method used ``float('inf')`` as the initial value for
+  ``shortest_session_seconds``, which is not valid JSON. Changed to ``None``
+  so serialisation succeeds when no session has completed yet.
+
+- **``wait_for()`` future not bound to running loop**: ``wait_for()`` created a
+  bare ``asyncio.Future()`` rather than
+  ``asyncio.get_running_loop().create_future()``, which could bind the future to
+  a different loop in multi-loop test setups.
+
+- **Reservation temperature validation was US-only**: ``build_reservation_entry``
+  validated set-point temperatures against hardcoded Fahrenheit bounds (95–150 °F)
+  regardless of the active unit system. Validation now uses the current unit system
+  context: 35–65 °C in metric mode, 95–150 °F in US mode. Celsius users previously
+  received spurious ``ValueError`` rejections for valid temperatures.
+
+- **Malformed reservation data silently dropped**: ``build_reservation_entry`` now
+  logs a warning when reservation hex data contains unexpected trailing bytes
+  instead of silently dropping partial entries.
+
+- **Unknown ``PeriodicRequestType`` silently ignored**: The periodic-request handler
+  now logs an error and breaks when it encounters an unknown request type instead of
+  doing nothing.
+
+- **Memory leak in device info cache**: ``get_all_cached()`` only filtered expired
+  entries from its return value but left them in the cache dictionary. Expired
+  entries are now evicted during ``get_all_cached()`` to prevent unbounded growth.
+
+Version 8.0.0 (2026-05-13)
+===========================
 
 Bug Fixes
 ---------

scripts/bump_version.py

@@ -22,6 +22,8 @@
 import re
 import subprocess
 import sys
+from datetime import date
+from pathlib import Path
 
 
 def run_git_command(args: list) -> str:
@@ -140,6 +142,75 @@ def check_working_directory_clean() -> None:
         sys.exit(1)
 
 
+def update_changelog(version: str) -> None:
+    """Insert a version heading into CHANGELOG.rst below the Unreleased section.
+
+    Transforms:
+
+        Unreleased
+        ==========
+
+        <content...>
+
+    into:
+
+        Unreleased
+        ==========
+
+        Version X.Y.Z (YYYY-MM-DD)
+        ===========================
+
+        <content...>
+    """
+    changelog_path = Path("CHANGELOG.rst")
+    if not changelog_path.exists():
+        print("Warning: CHANGELOG.rst not found, skipping changelog update.")
+        return
+
+    content = changelog_path.read_text(encoding="utf-8")
+
+    heading = f"Version {version} ({date.today().isoformat()})"
+    underline = "=" * len(heading)
+    version_block = f"{heading}\n{underline}\n"
+
+    # Match "Unreleased\n==========\n" (any number of = signs) followed by
+    # one or more blank lines, then insert the version block after them.
+    pattern = re.compile(
+        r"(Unreleased\n=+\n)"  # group 1: Unreleased heading
+        r"(\n+)",              # group 2: blank line(s) separator
+        re.MULTILINE,
+    )
+
+    match = pattern.search(content)
+    if not match:
+        print(
+            "Warning: Could not find 'Unreleased' section in CHANGELOG.rst. "
+            "Skipping changelog update.",
+            file=sys.stderr,
+        )
+        return
+
+    # Insert the version block after the blank lines that follow "Unreleased"
+    new_content = (
+        content[: match.end()]
+        + version_block
+        + "\n"
+        + content[match.end() :]
+    )
+
+    changelog_path.write_text(new_content, encoding="utf-8")
+    print(f"[OK] Updated CHANGELOG.rst with {heading}")
+
+
+def commit_changelog(version: str) -> None:
+    """Stage and commit the CHANGELOG.rst update."""
+    run_git_command(["add", "CHANGELOG.rst"])
+    run_git_command(
+        ["commit", "-m", f"Update changelog for v{version}"]
+    )
+    print("[OK] Committed changelog update")
+
+
 def create_tag(version: str, message: str = None) -> None:
     """Create a git tag for the version."""
     tag_name = f"v{version}"
@@ -223,13 +294,19 @@ def main() -> None:
     # Validate version progression
     validate_version_progression(current_version, new_version)
 
+    # Update CHANGELOG.rst and commit, then create the tag
+    print("\nUpdating CHANGELOG.rst...")
+    update_changelog(new_version)
+    commit_changelog(new_version)
+
     # Create the tag
     print(f"\nCreating tag v{new_version}...")
     create_tag(new_version)
 
     print("\n[OK] Version bump complete!")
     print("\nNext steps:")
     print(f"  1. Push the tag:    git push origin v{new_version}")
+    print("     (also push the changelog commit: git push origin HEAD)")
     print("  2. Build release:   make build")
     print("  3. Test on TestPyPI: make publish-test")
     print("  4. Publish to PyPI:  make publish")

src/nwp500/device_info_cache.py

@@ -137,12 +137,15 @@ async def get_all_cached(self) -> dict[str, DeviceFeature]:
             Dictionary mapping MAC addresses to DeviceFeature objects
         """
         async with self._lock:
-            # Filter out expired entries
-            return {
-                mac: features
-                for mac, (features, timestamp) in self._cache.items()
-                if not self.is_expired(timestamp)
-            }
+            # Filter out expired entries and purge them from cache
+            expired_keys = [
+                mac
+                for mac, (_, timestamp) in self._cache.items()
+                if self.is_expired(timestamp)
+            ]
+            for mac in expired_keys:
+                del self._cache[mac]
+            return {mac: features for mac, (features, _) in self._cache.items()}
 
     async def get_cache_info(
         self,

src/nwp500/encoding.py

@@ -8,11 +8,14 @@
 
 from __future__ import annotations
 
+import logging
 from collections.abc import Iterable
 from numbers import Real
 
 from .exceptions import ParameterValidationError, RangeValidationError
 
+_logger = logging.getLogger(__name__)
+
 # MGPP Week Bitfield Encoding (from NaviLink APK KDEnum.MgppReservationWeek).
 # Uses a single byte where bits 1-7 represent days; bit 0 is unused.
 #
@@ -342,14 +345,18 @@ def decode_reservation_hex(hex_string: str) -> list[dict[str, int]]:
     data = bytes.fromhex(hex_string)
     reservations = []
 
+    if len(data) % 6 != 0:
+        _logger.warning(
+            "Reservation hex data length %d is not a multiple of 6; "
+            "trailing %d bytes will be ignored",
+            len(data),
+            len(data) % 6,
+        )
+
     # Process 6 bytes at a time
-    for i in range(0, len(data), 6):
+    for i in range(0, len(data) - (len(data) % 6), 6):
         chunk = data[i : i + 6]
 
-        # Ensure we have a full 6-byte entry
-        if len(chunk) != 6:
-            break
-
         # Skip empty entries (all zeros)
         if all(b == 0 for b in chunk):
             continue
@@ -425,11 +432,26 @@ def build_reservation_entry(
     """
     # Import here to avoid circular import
     from .models import preferred_to_half_celsius
+    from .unit_system import get_unit_system
+
+    # Read unit system once to keep min/max bounds consistent
+    unit_system = get_unit_system()
 
     # Use device-provided limits if available, otherwise use defaults
-    # Defaults are conservative: 95°F / 35°C minimum, 150°F / 65°C maximum
-    min_temp = temperature_min if temperature_min is not None else 95
-    max_temp = temperature_max if temperature_max is not None else 150
+    # in the user's preferred unit system.
+    if temperature_min is not None:
+        min_temp = temperature_min
+    elif unit_system == "metric":
+        min_temp = 35.0  # ~35°C
+    else:
+        min_temp = 95.0  # 95°F
+
+    if temperature_max is not None:
+        max_temp = temperature_max
+    elif unit_system == "metric":
+        max_temp = 65.0  # ~65°C
+    else:
+        max_temp = 150.0  # 150°F
 
     if not 0 <= hour <= 23:
         raise RangeValidationError(

src/nwp500/events.py

@@ -396,7 +396,7 @@ async def wait_for(
             current_temp = temperature_event.new_temperature
         """
         future: asyncio.Future[tuple[tuple[Any, ...], dict[str, Any]]] = (
-            asyncio.Future()
+            asyncio.get_running_loop().create_future()
         )
 
         def handler(*args: Any, **kwargs: Any) -> None:

src/nwp500/mqtt/client.py

@@ -371,7 +371,10 @@ async def _active_reconnect(self) -> None:
         reconnect instead of passively waiting for AWS IoT SDK.
 
         Note: This creates a new connection while preserving subscriptions
-        and configuration.
+        and configuration. The old connection is closed first to prevent
+        its SDK auto-reconnect from creating a competing connection with
+        the same client ID (which causes the broker to kick one off,
+        leading to an infinite connect/disconnect loop).
         """
         if self._connected:
             _logger.debug("Already connected, skipping reconnection")
@@ -385,12 +388,15 @@ async def _active_reconnect(self) -> None:
 
             # If we have a connection manager, try to reconnect using it
             if self._connection_manager:
-                # The connection might be in a bad state, so we need to
-                # recreate the underlying connection
+                # Close old connection to stop SDK auto-reconnect and
+                # prevent two connections with the same client ID.
                 _logger.debug("Recreating MQTT connection...")
+                try:
+                    await self._connection_manager.close()
+                except (AwsCrtError, RuntimeError) as e:
+                    _logger.debug(f"Old connection cleanup (benign): {e}")
 
                 # Create a new connection manager with same config
-                old_connection_manager = self._connection_manager
                 self._connection_manager = MqttConnection(
                     config=self.config,
                     auth_client=self._auth_client,
@@ -415,9 +421,6 @@ async def _active_reconnect(self) -> None:
 
                     _logger.info("Active reconnection successful")
                 else:
-                    # Restore old connection manager and connection reference
-                    self._connection_manager = old_connection_manager
-                    self._connection = old_connection_manager.connection
                     _logger.warning("Active reconnection failed")
             else:
                 _logger.warning(
@@ -458,8 +461,7 @@ async def _deep_reconnect(self) -> None:
             if self._connection_manager:
                 _logger.debug("Cleaning up old connection...")
                 try:
-                    if self._connection_manager.is_connected:
-                        await self._connection_manager.disconnect()
+                    await self._connection_manager.close()
                 except (AwsCrtError, RuntimeError) as e:
                     # Expected: connection already dead or in bad state
                     _logger.debug(f"Error during cleanup: {e} (expected)")
@@ -1294,14 +1296,20 @@ async def ensure_device_info_cached(
             return True
 
         # Not cached, request and wait
-        future: asyncio.Future[DeviceFeature] = (
-            asyncio.get_running_loop().create_future()
-        )
+        loop = asyncio.get_running_loop()
+        future: asyncio.Future[DeviceFeature] = loop.create_future()
 
         def on_feature(feature: DeviceFeature) -> None:
-            if not future.done():
-                _logger.info(f"Device feature received for {redacted_mac}")
-                future.set_result(feature)
+            # Called from AWS SDK thread — schedule onto the event loop
+            # thread-safely. The done() check is inside the scheduled
+            # callback so it runs on the event loop thread, eliminating
+            # the race between the check and set_result.
+            def _set_result() -> None:
+                if not future.done():
+                    _logger.info(f"Device feature received for {redacted_mac}")
+                    future.set_result(feature)
+
+            loop.call_soon_threadsafe(_set_result)
 
         _logger.info(f"Ensuring device info cached for {redacted_mac}")
         await self.subscribe_device_feature(device, on_feature)