feat(ble): process-global per-MAC connection lock with contention WARNING

custom_components/opendisplay/__init__.py

@@ -33,6 +33,7 @@
 from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH
 from homeassistant.helpers.typing import ConfigType
 
+from .ble_lock import async_get_ble_lock, ble_connection
 from .const import CONF_CACHED_STATE, CONF_ENCRYPTION_KEY, DOMAIN, SETUP_DEADLINE_S
 from .coordinator import OpenDisplayCoordinator
 from .delivery import DeliveryManager
@@ -70,12 +71,15 @@ class OpenDisplayRuntimeData:
     is_flex: bool
     # Resolved deep-sleep behavior (options + device power config).
     sleep_profile: SleepProfile
+    # Serializes every BLE connection to this tag. Process-global per-MAC
+    # (ble_lock.py), so the same lock object is shared across every connect site
+    # and survives entry reloads. The device exposes a single BLE link with no
+    # per-address lock in the library, so drawcustom/upload_image, LED, buzzer
+    # and OTA must not open overlapping connections or they race and surface a
+    # confusing upload_error. Required (no default) so a future construction
+    # can't silently mint a private, non-shared lock and recreate this bug.
+    ble_lock: asyncio.Lock
     upload_task: asyncio.Task | None = None
-    # Serializes every BLE connection to this tag (one config entry == one MAC).
-    # The device exposes a single BLE link with no per-address lock in the
-    # library, so drawcustom/upload_image, LED, buzzer and OTA must not open
-    # overlapping connections or they race and surface a confusing upload_error.
-    ble_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
     # Tracks the last uploaded frame + etag for differential partial updates
     # (0x76). Replaced with a fresh instance on every full/fast refresh so the
     # next partial diffs against the frame actually on the panel.
@@ -235,7 +239,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: OpenDisplayConfigEntry)
             # wedged BLE link can't stall setup forever; a breach is treated like
             # any other connect failure below (sleepy-cache fallback / retry).
             async with asyncio.timeout(SETUP_DEADLINE_S):
-                async with OpenDisplayDevice(
+                async with ble_connection(address, "setup interrogation"), OpenDisplayDevice(
                     mac_address=address,
                     ble_device=ble_device,
                     encryption_key=encryption_key,
@@ -308,6 +312,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: OpenDisplayConfigEntry)
         device_config=device_config,
         is_flex=is_flex,
         sleep_profile=profile,
+        ble_lock=async_get_ble_lock(address),
         config_resync_pending=from_cache,
     )
 

custom_components/opendisplay/ble_lock.py

@@ -0,0 +1,78 @@
+"""Process-global per-MAC BLE connection lock for OpenDisplay tags.
+
+An OpenDisplay tag exposes a single BLE link and the library holds no
+per-address lock, so every host-side connect to a given MAC must be serialized
+or overlapping connects race and wedge BlueZ (the dongle-only config-read
+investigation, 2026-07-16). The lock has to be *process-global and keyed by
+MAC* rather than per config entry, because two of the connect sites run before
+(or without) a config entry: the config-flow probe and the entry-setup
+interrogation. A per-entry lock leaves those two unguarded against each other
+and against a reloading entry.
+
+``WeakValueDictionary`` here is **load-bearing, not hygiene**: an
+``asyncio.Lock`` binds to the event loop that first acquires it, and
+pytest-asyncio spins up a fresh loop per test while the test modules reuse one
+``ADDRESS`` constant. A plain ``dict`` would cache a lock bound to a now-dead
+loop and raise ``RuntimeError`` on the next test's acquire. Weak references let
+the entry drop the moment nothing holds the lock, so each test's first
+``async_get_ble_lock`` mints a lock bound to that test's live loop. In
+production the same weakness is harmless: a lock lingers only while a connect
+holds it and is otherwise collected.
+"""
+
+import asyncio
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+import logging
+from weakref import WeakValueDictionary
+
+from homeassistant.helpers.device_registry import format_mac
+
+_LOGGER = logging.getLogger(__name__)
+
+# Normalized MAC -> lock serializing every BLE connect to that tag. Weak values
+# so a lock is collected once nothing references it (see module docstring).
+_LOCKS: WeakValueDictionary[str, asyncio.Lock] = WeakValueDictionary()
+# Normalized MAC -> purpose string of the operation currently holding the lock,
+# used only to name the holder in the contention WARNING.
+_HOLDERS: dict[str, str] = {}
+
+
+def async_get_ble_lock(address: str) -> asyncio.Lock:
+    """Return the shared BLE lock for ``address``, creating it on first use.
+
+    ``address`` is normalized with ``format_mac`` because config_flow keys off
+    raw discovery addresses while everything else uses ``entry.unique_id``; both
+    must map to the same lock. There is no ``await`` between the lookup and the
+    insert, so this is atomic on the event loop.
+    """
+    key = format_mac(address)
+    if (lock := _LOCKS.get(key)) is None:
+        _LOCKS[key] = lock = asyncio.Lock()
+    return lock
+
+
+@asynccontextmanager
+async def ble_connection(address: str, purpose: str) -> AsyncIterator[None]:
+    """Hold the shared per-MAC BLE lock for the duration of a connection.
+
+    Emits a WARNING (before awaiting the lock) when the link is already held,
+    naming both the waiting ``purpose`` and the current holder, so overlapping
+    connect attempts on the same tag are diagnosable.
+    """
+    key = format_mac(address)
+    lock = async_get_ble_lock(address)
+    if lock.locked():
+        _LOGGER.warning(
+            "%s: BLE connection for %s requested while %s holds the device's "
+            "single BLE link; waiting for it to finish",
+            address,
+            purpose,
+            _HOLDERS.get(key, "another operation"),
+        )
+    async with lock:
+        _HOLDERS[key] = purpose
+        try:
+            yield
+        finally:
+            _HOLDERS.pop(key, None)

custom_components/opendisplay/config_flow.py

@@ -39,6 +39,7 @@
     SelectSelectorMode,
 )
 
+from .ble_lock import ble_connection
 from .const import (
     CONF_BLOCKS_PER_ACK,
     CONF_ENCRYPTION_KEY,
@@ -156,7 +157,9 @@ async def _async_test_connection(
         # maps to "cannot_connect"; AuthenticationRequiredError still propagates.
         try:
             async with asyncio.timeout(CONNECT_PROBE_DEADLINE_S):
-                async with OpenDisplayDevice(
+                async with ble_connection(
+                    address, "connection probe (config flow)"
+                ), OpenDisplayDevice(
                     mac_address=address,
                     ble_device=ble_device,
                     encryption_key=encryption_key,

custom_components/opendisplay/delivery.py

@@ -44,6 +44,7 @@
 from homeassistant.helpers.dispatcher import async_dispatcher_send
 from homeassistant.helpers.event import async_call_later
 
+from .ble_lock import ble_connection
 from .const import (
     CONF_BLOCKS_PER_ACK,
     CONF_ENCRYPTION_KEY,
@@ -307,7 +308,7 @@ async def _drain_once(self) -> None:
             return
 
         runtime = self._entry.runtime_data
-        async with runtime.ble_lock:
+        async with ble_connection(self._address, "queued content delivery"):
             ble_device = async_ble_device_from_address(
                 self._hass, self._address, connectable=True
             )

custom_components/opendisplay/services.py

@@ -61,6 +61,7 @@
 if TYPE_CHECKING:
     from . import OpenDisplayConfigEntry
 
+from .ble_lock import ble_connection
 from .const import (
     CONF_BLOCKS_PER_ACK,
     CONF_ENCRYPTION_KEY,
@@ -397,15 +398,15 @@ async def _async_connect_and_run(
         # and the library has no per-address lock, so overlapping connections
         # (two automations, or a drawcustom racing an LED/buzzer/upload on the
         # same MAC) fail with a confusing upload_error. The lock is held for the
-        # full connection lifetime and released on error. Held per config entry,
-        # i.e. per MAC, so different tags are not serialized against each other.
+        # full connection lifetime and released on error. Keyed per MAC, so
+        # different tags are not serialized against each other.
         # Same wall-clock ceiling as the queued-delivery drain: without it a
-        # wedged transfer would hold ble_lock forever and block every later
+        # wedged transfer would hold the lock forever and block every later
         # operation on this MAC (the library's per-read timeouts bound normal
         # failures, but not adversarial/buggy-firmware frame streams).
         async with (
             asyncio.timeout(DELIVERY_DEADLINE_S),
-            entry.runtime_data.ble_lock,
+            ble_connection(address, "service call (upload/drawcustom/LED/buzzer)"),
             OpenDisplayDevice(
                 mac_address=address,
                 ble_device=ble_device,

custom_components/opendisplay/update.py

@@ -32,6 +32,7 @@
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
 from . import OpenDisplayConfigEntry, _get_encryption_key
+from .ble_lock import ble_connection
 from .const import DOMAIN
 from .entity import OpenDisplayEntity
 
@@ -245,7 +246,7 @@ def _on_log(msg: str) -> None:
             # half-connecting can't hold the lock forever (OTA_INSTALL_DEADLINE_S).
             async with (
                 asyncio.timeout(OTA_INSTALL_DEADLINE_S) as ota_deadline,
-                self._entry.runtime_data.ble_lock,
+                ble_connection(self._ble_address, "firmware update (OTA)"),
             ):
                 # Only EFR32BG22 (Silabs AppLoader) is flashed over BLE here — see
                 # _OTA_INSTALL_IC_TYPES. The device is either in app mode (and needs

tests/test_ble_lock.py

@@ -0,0 +1,128 @@
+"""Unit tests for the process-global per-MAC BLE connection lock.
+
+Exercises the registry keying (case-normalization, distinct MACs), the
+contention WARNING, holder bookkeeping, and the ``WeakValueDictionary``
+collection that keeps a lock from outliving the loop that bound it. These tests
+run on pytest-asyncio's per-test event loop while reusing one ``ADDRESS`` — the
+exact shape that would raise ``RuntimeError`` if the registry cached a
+loop-bound lock in a plain dict.
+"""
+
+import asyncio
+import gc
+import logging
+
+import pytest
+
+from homeassistant.helpers.device_registry import format_mac
+
+from custom_components.opendisplay.ble_lock import (
+    _HOLDERS,
+    _LOCKS,
+    async_get_ble_lock,
+    ble_connection,
+)
+
+ADDRESS = "AA:BB:CC:DD:EE:FF"
+_LOGGER_NAME = "custom_components.opendisplay.ble_lock"
+
+
+@pytest.mark.asyncio
+async def test_case_variant_addresses_share_one_lock():
+    """An upper- and lower-cased MAC resolve to the same lock object."""
+    upper = async_get_ble_lock("AA:BB:CC:DD:EE:FF")
+    lower = async_get_ble_lock("aa:bb:cc:dd:ee:ff")
+    assert upper is lower
+
+
+@pytest.mark.asyncio
+async def test_distinct_macs_get_distinct_locks():
+    """Different MACs are not serialized against each other."""
+    a = async_get_ble_lock("AA:BB:CC:DD:EE:FF")
+    b = async_get_ble_lock("11:22:33:44:55:66")
+    assert a is not b
+
+
+@pytest.mark.asyncio
+async def test_contended_connection_serializes_and_warns(caplog):
+    """A second connect on a held link warns (naming both ops) and waits."""
+    caplog.set_level(logging.WARNING, logger=_LOGGER_NAME)
+    order: list[str] = []
+    release_a = asyncio.Event()
+    a_holding = asyncio.Event()
+
+    async def op_a() -> None:
+        async with ble_connection(ADDRESS, "op-a"):
+            order.append("a-enter")
+            a_holding.set()
+            await release_a.wait()
+            order.append("a-exit")
+
+    async def op_b() -> None:
+        await a_holding.wait()
+        async with ble_connection(ADDRESS, "op-b"):
+            order.append("b-enter")
+
+    task_a = asyncio.create_task(op_a())
+    await a_holding.wait()
+    task_b = asyncio.create_task(op_b())
+    # Let B reach (and block on) the contended acquire while A still holds.
+    for _ in range(5):
+        await asyncio.sleep(0)
+    release_a.set()
+    await asyncio.gather(task_a, task_b)
+
+    # B's body runs only after A fully exits.
+    assert order == ["a-enter", "a-exit", "b-enter"]
+
+    warnings = [
+        r for r in caplog.records
+        if r.name == _LOGGER_NAME and r.levelno == logging.WARNING
+    ]
+    assert len(warnings) == 1
+    message = warnings[0].getMessage()
+    assert ADDRESS in message  # waiter's address
+    assert "op-b" in message   # the waiting purpose
+    assert "op-a" in message   # the current holder
+
+
+@pytest.mark.asyncio
+async def test_uncontended_connection_emits_no_warning(caplog):
+    """A connect on a free link logs nothing."""
+    caplog.set_level(logging.WARNING, logger=_LOGGER_NAME)
+    async with ble_connection(ADDRESS, "solo"):
+        pass
+
+    warnings = [
+        r for r in caplog.records
+        if r.name == _LOGGER_NAME and r.levelno == logging.WARNING
+    ]
+    assert warnings == []
+
+
+@pytest.mark.asyncio
+async def test_holder_cleared_after_exit():
+    """The holder entry is set while held and popped in the finally."""
+    key = format_mac(ADDRESS)
+    async with ble_connection(ADDRESS, "op"):
+        assert _HOLDERS[key] == "op"
+    assert key not in _HOLDERS
+
+
+@pytest.mark.asyncio
+async def test_unreferenced_lock_is_collected():
+    """Dropping the last reference lets the WeakValueDictionary evict the lock.
+
+    This is what lets the next test's first acquire mint a lock bound to that
+    test's live loop instead of resurrecting a dead-loop one.
+    """
+    key = format_mac(ADDRESS)
+    lock = async_get_ble_lock(ADDRESS)
+    assert _LOCKS.get(key) is lock
+    del lock
+    gc.collect()
+    assert key not in _LOCKS
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))