diff --git a/custom_components/nestquest/sweep.py b/custom_components/nestquest/sweep.py index a36cdf4..e60ab72 100644 --- a/custom_components/nestquest/sweep.py +++ b/custom_components/nestquest/sweep.py @@ -21,6 +21,7 @@ skip it via the no-completion-event rule. """ from __future__ import annotations +import asyncio import datetime import logging from zoneinfo import ZoneInfo @@ -37,6 +38,25 @@ LOGGER = logging.getLogger(__name__) #: meta_state key holding the HA-local date the sweep last ran. SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date" +#: One asyncio.Lock per (connection wrapper, running loop), mirroring +#: the migration runner's pattern: the whole watermark read → query → +#: fire → write span is one critical section, so an overlapping startup +#: sweep and rollover sweep queue instead of double-announcing. The +#: loser re-reads the watermark INSIDE the lock, sees the winner's +#: date, and fires nothing. +_SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {} + + +def _sweep_lock(database: NestQuestDatabase) -> asyncio.Lock: + """Return the sweep lock bound to ``database``'s running loop.""" + loop = asyncio.get_running_loop() + key = (id(database), id(loop)) + lock = _SWEEP_LOCKS.get(key) + if lock is None: + lock = asyncio.Lock() + _SWEEP_LOCKS[key] = lock + return lock + #: UTC ISO-8601 timestamp policy shared with the DAO layers. _UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00" @@ -60,6 +80,12 @@ async def run_missed_sweep( from ``hass.config.time_zone``. Returns the number of events fired. Fires nothing and returns 0 when the watermark already covers ``today`` (a same-night rerun). + + The whole watermark read → query → fire → write span runs inside + the per-database sweep lock: an overlapping startup sweep and a + rollover sweep queue instead of interleaving, and the loser + re-reads the watermark inside the lock — sees the winner's date — + and announces nothing twice. """ today_date = ( today @@ -68,35 +94,38 @@ async def run_missed_sweep( ) today_iso = today_date.isoformat() dao = MetaStateDao(database) - watermark = await dao.get(SWEEP_WATERMARK_KEY) - if watermark == today_iso: - LOGGER.debug("NestQuest missed sweep already ran for %s", today_iso) - return 0 + async with _sweep_lock(database): + watermark = await dao.get(SWEEP_WATERMARK_KEY) + if watermark == today_iso: + LOGGER.debug( + "NestQuest missed sweep already ran for %s", today_iso + ) + return 0 - instances_dao = QuestInstancesDao(database) - rows = await instances_dao.list_missed_unswept(today_iso, watermark) - children_dao = ChildrenDao(database) - definitions_dao = QuestDefinitionsDao(database) - now_stamp = _now_stamp() - fired = 0 - for instance in rows: - child = await children_dao.get(instance.child_id) - definition = await definitions_dao.get(instance.definition_id) - hass.bus.async_fire( - EVENT_QUEST_MISSED, - { - "child_id": instance.child_id, - "child_name": child.display_name if child else None, - "instance_id": instance.id, - "quest_title": definition.title if definition else None, - "window": instance.window, - "due_date": instance.due_date, - "due_time": instance.due_time, - "occurred_at": now_stamp, - }, - ) - fired += 1 - await dao.set(SWEEP_WATERMARK_KEY, today_iso) + instances_dao = QuestInstancesDao(database) + rows = await instances_dao.list_missed_unswept(today_iso, watermark) + children_dao = ChildrenDao(database) + definitions_dao = QuestDefinitionsDao(database) + now_stamp = _now_stamp() + fired = 0 + for instance in rows: + child = await children_dao.get(instance.child_id) + definition = await definitions_dao.get(instance.definition_id) + hass.bus.async_fire( + EVENT_QUEST_MISSED, + { + "child_id": instance.child_id, + "child_name": child.display_name if child else None, + "instance_id": instance.id, + "quest_title": definition.title if definition else None, + "window": instance.window, + "due_date": instance.due_date, + "due_time": instance.due_time, + "occurred_at": now_stamp, + }, + ) + fired += 1 + await dao.set(SWEEP_WATERMARK_KEY, today_iso) LOGGER.info( "NestQuest missed sweep for %s fired %d event(s)", today_iso, fired ) diff --git a/tests/test_sweep.py b/tests/test_sweep.py index 17ee530..c76ac81 100644 --- a/tests/test_sweep.py +++ b/tests/test_sweep.py @@ -237,3 +237,23 @@ async def test_v6_database_migrates_to_v7_meta_state( async def _count_events(database) -> int: rows = await database.fetch_all("SELECT COUNT(*) FROM completion_events") return rows[0][0] + + +async def test_overlapping_sweeps_never_double_announce( + hass, make_entry +) -> None: + """A startup sweep and a rollover sweep racing on the same night + queue on the sweep lock: the winner announces, the loser re-reads + the watermark inside the lock and fires nothing.""" + import asyncio + + entry, _ada, _bo = await _seed(hass, make_entry) + database = entry.runtime_data.database + tomorrow = datetime.date.today() + datetime.timedelta(days=1) + + first, second = await asyncio.gather( + run_missed_sweep(hass, database, today=tomorrow), + run_missed_sweep(hass, database, today=tomorrow), + ) + assert first + second == 2, "each instance announced exactly once" + assert len(hass.bus.fired(EVENT_QUEST_MISSED)) == 2