diff --git a/custom_components/opendisplay/__init__.py b/custom_components/opendisplay/__init__.py index c4d777a..34fccf8 100644 --- a/custom_components/opendisplay/__init__.py +++ b/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, ) diff --git a/custom_components/opendisplay/ble_lock.py b/custom_components/opendisplay/ble_lock.py new file mode 100644 index 0000000..6258dd7 --- /dev/null +++ b/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) diff --git a/custom_components/opendisplay/config_flow.py b/custom_components/opendisplay/config_flow.py index ae229c2..f11ab71 100644 --- a/custom_components/opendisplay/config_flow.py +++ b/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, diff --git a/custom_components/opendisplay/delivery.py b/custom_components/opendisplay/delivery.py index 8378b89..4dc9492 100644 --- a/custom_components/opendisplay/delivery.py +++ b/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 ) diff --git a/custom_components/opendisplay/services.py b/custom_components/opendisplay/services.py index b8dcd0a..eab5307 100644 --- a/custom_components/opendisplay/services.py +++ b/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, diff --git a/custom_components/opendisplay/update.py b/custom_components/opendisplay/update.py index 65dd156..56f259c 100644 --- a/custom_components/opendisplay/update.py +++ b/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 diff --git a/tests/test_ble_lock.py b/tests/test_ble_lock.py new file mode 100644 index 0000000..59e5424 --- /dev/null +++ b/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"])) diff --git a/tests/test_delivery.py b/tests/test_delivery.py index 994d752..3e550e8 100644 --- a/tests/test_delivery.py +++ b/tests/test_delivery.py @@ -6,6 +6,7 @@ """ import asyncio +import logging from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -15,6 +16,7 @@ from homeassistant.exceptions import HomeAssistantError from custom_components.opendisplay import delivery as delivery_mod +from custom_components.opendisplay.ble_lock import async_get_ble_lock, ble_connection from custom_components.opendisplay.const import ( CONF_BLOCKS_PER_ACK, CONF_MAX_QUEUE_SIZE, @@ -52,7 +54,6 @@ def _make_env(profile=None, entry_data=None, last_seen=None, options=None): coordinator=coordinator, sleep_profile=profile, device_config=MagicMock(), - ble_lock=asyncio.Lock(), config_resync_pending=False, firmware=None, is_flex=False, @@ -465,5 +466,93 @@ async def test_drain_passes_state_and_refresh_mode(): assert call.kwargs["refresh_mode"] is RefreshMode.PARTIAL +@pytest.mark.asyncio +async def test_drain_holds_registry_lock_during_connection(): + """The drain body runs while the process-global per-MAC lock is held.""" + hass, entry, _ = _make_env() + device = MagicMock() + device.upload_prepared_image = AsyncMock() + held: dict[str, bool] = {} + + def _factory(**kwargs): + class _Ctx: + async def __aenter__(self): + held["locked"] = async_get_ble_lock(ADDRESS).locked() + return device + + async def __aexit__(self, *exc): + return False + + return _Ctx() + + with ( + patch.object(delivery_mod, "async_call_later", return_value=MagicMock()), + patch.object(delivery_mod, "async_dispatcher_send"), + patch.object(delivery_mod, "async_ble_device_from_address", return_value=MagicMock()), + patch.object(delivery_mod, "OpenDisplayDevice", side_effect=_factory), + ): + mgr = DeliveryManager(hass, entry) + _submit(mgr) + await mgr._deliver() + + assert held["locked"] is True + + +@pytest.mark.asyncio +async def test_drain_waits_for_preheld_registry_lock(caplog): + """A drain queued behind a pre-held registry lock warns and waits.""" + caplog.set_level(logging.WARNING, logger="custom_components.opendisplay.ble_lock") + hass, entry, _ = _make_env() + device = MagicMock() + device.upload_prepared_image = AsyncMock() + order: list[str] = [] + release = asyncio.Event() + + async def _hold() -> None: + async with ble_connection(ADDRESS, "external holder"): + order.append("holder-enter") + await release.wait() + order.append("holder-exit") + + def _factory(**kwargs): + class _Ctx: + async def __aenter__(self): + order.append("drain-connect") + return device + + async def __aexit__(self, *exc): + return False + + return _Ctx() + + with ( + patch.object(delivery_mod, "async_call_later", return_value=MagicMock()), + patch.object(delivery_mod, "async_dispatcher_send"), + patch.object(delivery_mod, "async_ble_device_from_address", return_value=MagicMock()), + patch.object(delivery_mod, "OpenDisplayDevice", side_effect=_factory), + ): + mgr = DeliveryManager(hass, entry) + _submit(mgr) + holder = asyncio.create_task(_hold()) + while "holder-enter" not in order: + await asyncio.sleep(0) + drain = asyncio.create_task(mgr._deliver()) + # The drain must block on the held lock before it ever connects. + for _ in range(5): + await asyncio.sleep(0) + assert "drain-connect" not in order + release.set() + await asyncio.gather(holder, drain) + + assert order == ["holder-enter", "holder-exit", "drain-connect"] + device.upload_prepared_image.assert_awaited_once() + warnings = [ + r for r in caplog.records + if r.name == "custom_components.opendisplay.ble_lock" + and r.levelno == logging.WARNING + ] + assert len(warnings) == 1 + + if __name__ == "__main__": raise SystemExit(pytest.main([__file__, "-v"])) diff --git a/tests/test_services.py b/tests/test_services.py index 2065e97..22fac50 100644 --- a/tests/test_services.py +++ b/tests/test_services.py @@ -7,6 +7,7 @@ """ import asyncio +import logging import time from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -16,6 +17,7 @@ import pytest from custom_components.opendisplay import services as services_mod +from custom_components.opendisplay.ble_lock import ble_connection from custom_components.opendisplay.const import ( CONF_BLOCKS_PER_ACK, CONF_MAX_QUEUE_SIZE, @@ -61,7 +63,6 @@ def _make_env(profile=None, last_seen=None, options=None): sleep_profile=profile, delivery=manager, device_config=None, - ble_lock=asyncio.Lock(), partial_state=MagicMock(), ) entry = MagicMock() @@ -326,5 +327,55 @@ async def test_live_send_passes_state_and_refresh_mode(): assert call.kwargs["refresh_mode"] is RefreshMode.FULL +@pytest.mark.asyncio +async def test_live_send_waits_for_preheld_registry_lock(caplog): + """A live send queued behind a pre-held per-MAC lock warns and waits.""" + caplog.set_level(logging.WARNING, logger="custom_components.opendisplay.ble_lock") + hass, entry, _, _ = _make_env(profile=_profile(sleep_mode="off"), last_seen=None) + device = MagicMock() + device.upload_prepared_image = AsyncMock() + order: list[str] = [] + release = asyncio.Event() + + async def _hold() -> None: + async with ble_connection(ADDRESS, "external holder"): + order.append("holder-enter") + await release.wait() + order.append("holder-exit") + + def _od_factory(**kwargs): + class _Ctx: + async def __aenter__(self): + order.append("send-connect") + return device + + async def __aexit__(self, *exc): + return False + + return _Ctx() + + p1, p2, p3, p4, p5 = _patches(MagicMock(side_effect=_od_factory)) + with p1, p2, p3, p4, p5: + holder = asyncio.create_task(_hold()) + while "holder-enter" not in order: + await asyncio.sleep(0) + send = asyncio.create_task(_send(hass, entry)) + # The send must block on the held lock before it ever connects. + for _ in range(5): + await asyncio.sleep(0) + assert "send-connect" not in order + release.set() + receipt, _ = await asyncio.gather(send, holder) + + assert order == ["holder-enter", "holder-exit", "send-connect"] + assert receipt.status == "delivered" + warnings = [ + r for r in caplog.records + if r.name == "custom_components.opendisplay.ble_lock" + and r.levelno == logging.WARNING + ] + assert len(warnings) == 1 + + if __name__ == "__main__": raise SystemExit(pytest.main([__file__, "-v"]))