Serialize the missed sweep's watermark claim

The whole watermark read-query-fire-write span runs under a
per-database sweep lock; a racing second sweep re-reads the watermark
inside the lock and announces nothing.

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,9 +94,12 @@ async def run_missed_sweep(
     )
     today_iso = today_date.isoformat()
     dao = MetaStateDao(database)
+    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)
+            LOGGER.debug(
+                "NestQuest missed sweep already ran for %s", today_iso
+            )
             return 0
 
         instances_dao = QuestInstancesDao(database)

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