fix(transport): close all BLE notification-watcher leak paths

src/opendisplay/device.py

@@ -9,7 +9,7 @@
 import logging
 import time
 from collections.abc import AsyncIterator, Awaitable, Callable
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, suppress
 from typing import TYPE_CHECKING, TypeVar, cast
 
 from epaper_dithering import ColorScheme, DitherMode, dither_image
@@ -581,18 +581,31 @@ async def __aenter__(self) -> OpenDisplayDevice:
 
         await self._conn.connect()
 
-        # Authenticate before any other commands if key provided
-        if self._encryption_key is not None:
-            await self.authenticate(self._encryption_key)
-
-        # Auto-interrogate if no config or capabilities provided
-        if self._config is None and self._capabilities is None:
-            _LOGGER.info("No config provided, auto-interrogating device")
-            await self.interrogate()
-
-        # Extract capabilities from config if available
-        if self._config and not self._capabilities:
-            self._capabilities = self._extract_capabilities_from_config()
+        # The link is now up with notifications registered. If any step below
+        # raises, __aexit__ will NOT run (Python only calls it when __aenter__
+        # returns) — so we must tear the link down here or leak the connection
+        # and its BlueZ notification watcher. BaseException, not Exception: the
+        # real triggers are AuthenticationRequiredError on a keyless probe AND
+        # CancelledError from an outer asyncio.timeout() (e.g. the HA config-flow
+        # connection probe).
+        try:
+            # Authenticate before any other commands if key provided
+            if self._encryption_key is not None:
+                await self.authenticate(self._encryption_key)
+
+            # Auto-interrogate if no config or capabilities provided
+            if self._config is None and self._capabilities is None:
+                _LOGGER.info("No config provided, auto-interrogating device")
+                await self.interrogate()
+
+            # Extract capabilities from config if available
+            if self._config and not self._capabilities:
+                self._capabilities = self._extract_capabilities_from_config()
+        except BaseException:
+            with suppress(Exception):
+                await self._conn.disconnect()
+            self._clear_session()
+            raise
 
         return self
 

src/opendisplay/transport/connection.py

@@ -194,6 +194,25 @@ async def _attempt_connect(self, *, use_services_cache: bool) -> None:
         # Start notifications
         await self._setup_notifications()
 
+    async def _stop_notifications(self) -> None:
+        """Best-effort release of the notification subscription (and its BlueZ
+        watcher) for the current client.
+
+        Called before every disconnect so the watcher is released even if the
+        subsequent ACL disconnect raises or is flaky. A disconnect that fails
+        while notifications are still registered is the watcher-leak precondition:
+        stopping notifications first makes nulling the client reference afterwards
+        safe. Never raises.
+        """
+        client = self._client
+        char = self._notification_characteristic
+        if client is None or char is None or not client.is_connected:
+            return
+        try:
+            await client.stop_notify(char)
+        except Exception as err:  # noqa: BLE001 - best-effort; link may be tearing down
+            _LOGGER.debug("stop_notify during teardown failed: %s", err)
+
     async def _clear_cache_and_drop(self) -> None:
         """Clear the proxy GATT cache on the current client, then disconnect it."""
         if self._client is None:
@@ -204,21 +223,31 @@ async def _clear_cache_and_drop(self) -> None:
                 await clear()  # pylint: disable=not-callable
             except Exception as err:  # noqa: BLE001 - best-effort
                 _LOGGER.debug("clear_cache during connect retry failed: %s", err)
+        # Release the notification watcher BEFORE disconnecting so a flaky
+        # disconnect can't strand it (the watcher-leak precondition). The retry
+        # loop relies on _client being None afterwards, so we always null below.
+        await self._stop_notifications()
         try:
             await self._client.disconnect()
         except Exception:  # noqa: BLE001
             pass
+        self._notification_characteristic = None
         self._client = None
 
     async def disconnect(self) -> None:
-        """Disconnect from device."""
+        """Disconnect from device, releasing notifications first."""
         if self._client and self._client.is_connected:
+            # Stop notifications before disconnecting so a disconnect that raises
+            # can't leave the BlueZ watcher registered (the watcher-leak that
+            # delivers stray/duplicate frames to a later connection).
+            await self._stop_notifications()
             try:
                 _LOGGER.debug("Disconnecting from %s", self.mac_address)
                 await self._client.disconnect()
             except Exception as e:
                 _LOGGER.warning("Error during disconnect: %s", e)
             finally:
+                self._notification_characteristic = None
                 self._client = None
 
     async def clear_cache(self) -> bool:

tests/unit/test_connection_watcher_leak.py

@@ -0,0 +1,144 @@
+"""Regression tests for BlueZ notification-watcher leaks.
+
+A watcher is registered by ``start_notify`` and only released when the client is
+disconnected (or ``stop_notify`` is called). Every code path that can reach a
+connected+subscribed client MUST guarantee a disconnect, or a stale watcher
+lingers and delivers stray/duplicate notifications to a later connection.
+
+Three leak paths, one test each:
+  * Path 1 — ``OpenDisplayDevice.__aenter__`` raises after ``connect()`` (e.g.
+    interrogate times out, or an outer ``asyncio.timeout`` cancels it). Because
+    ``__aexit__`` only runs when ``__aenter__`` returns, the connection must be
+    torn down inside ``__aenter__`` itself.
+  * Path 2 — ``_clear_cache_and_drop`` disconnect is flaky.
+  * Path 3 — the normal ``disconnect`` disconnect is flaky.
+Paths 2 & 3 release the watcher with an explicit ``stop_notify`` BEFORE the
+(possibly failing) disconnect, so nulling the client afterwards can't strand it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from bleak.exc import BleakError
+
+from opendisplay import OpenDisplayDevice
+from opendisplay.transport.connection import BLEConnection
+
+
+def _connected(client: object) -> BLEConnection:
+    conn = BLEConnection("AA:BB:CC:DD:EE:FF")
+    conn._client = client  # type: ignore[assignment]
+    conn._notification_characteristic = MagicMock()  # a live subscription
+    return conn
+
+
+# --- Path 1: __aenter__ must disconnect on any post-connect failure -----------
+
+
+@pytest.mark.asyncio
+async def test_aenter_disconnects_when_interrogate_raises() -> None:
+    """A failure after connect() (here interrogate) must disconnect the link;
+    otherwise __aexit__ never runs and the connection + watcher leak."""
+    device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF")
+    fake_conn = MagicMock()
+    fake_conn.connect = AsyncMock()
+    fake_conn.disconnect = AsyncMock()
+
+    class _Boom(Exception):
+        pass
+
+    with patch("opendisplay.device.BLEConnection", return_value=fake_conn):
+        device.interrogate = AsyncMock(side_effect=_Boom("no response"))  # type: ignore[method-assign]
+        with pytest.raises(_Boom):
+            await device.__aenter__()
+
+    fake_conn.disconnect.assert_awaited_once()  # link torn down on the failure path
+
+
+@pytest.mark.asyncio
+async def test_aenter_disconnects_on_cancellation() -> None:
+    """The real-world trigger: an outer asyncio.timeout() cancels the probe while
+    interrogate is in flight. CancelledError is a BaseException, so the guard must
+    catch BaseException — a plain 'except Exception' would leak here."""
+    device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF")
+    fake_conn = MagicMock()
+    fake_conn.connect = AsyncMock()
+    fake_conn.disconnect = AsyncMock()
+
+    with patch("opendisplay.device.BLEConnection", return_value=fake_conn):
+        device.interrogate = AsyncMock(side_effect=asyncio.CancelledError())  # type: ignore[method-assign]
+        with pytest.raises(asyncio.CancelledError):
+            await device.__aenter__()
+
+    fake_conn.disconnect.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_aenter_success_does_not_disconnect() -> None:
+    """The happy path must NOT disconnect — __aexit__ owns teardown there."""
+    device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF")
+    fake_conn = MagicMock()
+    fake_conn.connect = AsyncMock()
+    fake_conn.disconnect = AsyncMock()
+
+    with patch("opendisplay.device.BLEConnection", return_value=fake_conn):
+        device.interrogate = AsyncMock()  # type: ignore[method-assign]
+        result = await device.__aenter__()
+
+    assert result is device
+    fake_conn.disconnect.assert_not_awaited()
+
+
+# --- Path 3: disconnect() releases the watcher before disconnecting -----------
+
+
+@pytest.mark.asyncio
+async def test_disconnect_stops_notifications_first() -> None:
+    """disconnect() stops notifications before the ACL disconnect and clears state."""
+    client = AsyncMock(is_connected=True)
+    conn = _connected(client)
+    char = conn._notification_characteristic
+
+    await conn.disconnect()
+
+    client.stop_notify.assert_awaited_once_with(char)
+    client.disconnect.assert_awaited_once()
+    assert conn._client is None
+    assert conn._notification_characteristic is None
+
+
+@pytest.mark.asyncio
+async def test_disconnect_releases_watcher_even_if_disconnect_raises() -> None:
+    """A disconnect that raises must not strand the watcher: stop_notify already
+    released it, and disconnect() swallows the error rather than propagating."""
+    client = AsyncMock(is_connected=True)
+    client.disconnect = AsyncMock(side_effect=BleakError("link wedged"))
+    conn = _connected(client)
+    char = conn._notification_characteristic
+
+    await conn.disconnect()  # must not raise
+
+    client.stop_notify.assert_awaited_once_with(char)  # watcher released pre-disconnect
+    assert conn._client is None
+
+
+# --- Path 2: _clear_cache_and_drop releases the watcher before disconnecting ---
+
+
+@pytest.mark.asyncio
+async def test_clear_cache_and_drop_stops_notifications_first() -> None:
+    """The connect-retry drop path also releases the watcher before disconnecting."""
+    client = AsyncMock(is_connected=True)
+    client.clear_cache = AsyncMock()
+    conn = _connected(client)
+    char = conn._notification_characteristic
+
+    await conn._clear_cache_and_drop()
+
+    client.stop_notify.assert_awaited_once_with(char)
+    client.disconnect.assert_awaited_once()
+    assert conn._client is None
+    assert conn._notification_characteristic is None