Make the missed-sweep watermark claim safe across processes

The sweep guarded its announcement with a read-then-set watermark
inside a per-process asyncio lock, so two processes sharing one SQLite
file (the HA integration's rollover sweep and the API scheduler) could
both read the old watermark and both announce the same missed batch.

Add MetaStateDao.claim: a single conditional upsert
(INSERT ... ON CONFLICT DO UPDATE ... WHERE value IS NOT excluded.value)
whose rowcount says whether THIS caller changed the row — the
read-compare-write is one SQLite statement, atomic at the database
level, not just within one process. run_missed_sweep now builds the
events optimistically and gates the announcement on that claim: the
caller that advances the watermark returns the events, a loser returns
[]. The claim happens only after the events are built, so a failure
before it leaves the watermark unadvanced and a later run retries;
get/set are intact and the in-process sweep lock still serialises
same-process callers.

Tests prove two independent NestQuestDatabase wrappers (the
two-process shape) racing one seeded database yield exactly one
announcer, plus the unchanged same-night no-op and post-downtime
window-once semantics.

custom_components/nestquest/core/dao_meta.py

@@ -41,4 +41,38 @@ class MetaStateDao:
                     f"INSERT INTO {_TABLE} (key, value) VALUES (?, ?) "
                     "ON CONFLICT (key) DO UPDATE SET value = excluded.value",
                     (key, value),
-                )
+                )
+
+    async def claim(self, key: str, value: str) -> bool:
+        """Atomically set ``key`` to ``value`` unless it already is.
+
+        The single upsert statement carries a ``WHERE`` guard on the
+        conflict path (``t.value IS NOT excluded.value``), so the
+        read-compare-write is ONE SQLite statement — atomic at the
+        DATABASE level, not just within one process: two connections
+        (two processes sharing one SQLite file) cannot both observe
+        "not yet set" and both write, because the second writer's
+        statement re-evaluates the guard against the first writer's
+        committed value and matches zero rows.
+
+        Returns True only when THIS call changed the row: the first
+        claimant wins and every later claimant of the same value
+        returns False.  A DIFFERENT value always wins (an advance, e.g.
+        the sweep watermark moving to a new date) and re-arms the
+        same-value loser check for the new value.
+
+        Runs under the shared connection lock and inside one
+        transaction, like :meth:`set`; ``cursor.rowcount`` (1 for a
+        matched write, 0 for a guarded no-op) is captured by the
+        wrapper on the executor thread, so no sqlite3 object crosses
+        back to the event loop.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                result = await self._database.execute(
+                    f"INSERT INTO {_TABLE} (key, value) VALUES (?, ?) "
+                    "ON CONFLICT (key) DO UPDATE SET value = excluded.value "
+                    f"WHERE {_TABLE}.value IS NOT excluded.value",
+                    (key, value),
+                )
+        return result.rowcount == 1

custom_components/nestquest/core/sweep.py

@@ -26,12 +26,17 @@ later completed simply gains its completion event; history keeps both
 rows (append-only), and the sweep's later runs skip it via the
 no-completion-event rule.
 
-The whole watermark read → query → build → write span runs inside the
+The whole watermark read → query → build span runs inside the
 per-database sweep lock (one lock per connection wrapper and running
 loop, the settings store's pattern): 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.
+a rollover sweep of the SAME process queue instead of interleaving.
+The announce decision itself is the DATABASE-level claim
+(:meth:`MetaStateDao.claim`), so two PROCESSES sharing one SQLite file
+(the integration's rollover sweep and the API service's scheduler)
+cannot both announce the same missed batch either: both may read the
+old watermark and build the events, but the single conditional upsert
+lets exactly one of them win the claim — the loser returns [] and
+announces nothing.
 
 The caller supplies ``today`` as a plain :class:`datetime.date`: the
 household-local calendar date, read from ``hass.config.time_zone`` by
@@ -60,11 +65,11 @@ LOGGER = logging.getLogger(__name__)
 SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
 
 #: One asyncio.Lock per (connection wrapper, running loop), mirroring
-#: the settings store's pattern: the whole watermark read → query →
-#: build → 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.
+#: the settings store's pattern: the watermark read → query → build
+#: span is one critical section, so an overlapping startup sweep and
+#: rollover sweep of the SAME process queue instead of interleaving.
+#: Same-process callers are fully serialised by it; the cross-process
+#: case is closed by the database-level claim, not by this lock.
 _SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
 
 
@@ -99,15 +104,24 @@ async def run_missed_sweep(
     the caller decides how to publish them and announces nothing on
     its own.
 
-    The whole watermark read → query → build → 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.  Every event in one run shares one
+    The watermark read → query → build span runs inside the
+    per-database sweep lock, so overlapping callers of ONE process
+    queue instead of interleaving.  The announce decision is the
+    DATABASE-level claim (:meth:`MetaStateDao.claim` — a single
+    conditional upsert on the watermark, atomic across processes):
+    every caller builds the events optimistically, then only the
+    caller whose claim actually advanced the watermark returns them;
+    a loser (another process already announced for ``today``) returns
+    [] and announces nothing twice.  Every event in one run shares one
     ``occurred_at`` stamp (one run, one announcement moment), and the
     payload comes from the shared
     :func:`~.events._instance_payload` builder so the shape can never
     drift from the documented §3 contract.
+
+    Failure ordering: the watermark is advanced ONLY by the claim,
+    AFTER the events are fully built — a failure before the claim
+    leaves the watermark unadvanced, so a later run re-sweeps the same
+    window and nothing is lost.
     """
     today_iso = today.isoformat()
     dao = MetaStateDao(database)
@@ -134,10 +148,23 @@ async def run_missed_sweep(
                     ),
                 )
             )
-        await dao.set(SWEEP_WATERMARK_KEY, today_iso)
+        # The database-level claim gates the announcement: only the
+        # caller that actually advanced the watermark may publish the
+        # built events, so two processes racing one SQLite file cannot
+        # both announce the same missed batch.  A same-value loser's
+        # claim matches zero rows and this run announces nothing.
+        claimed = await dao.claim(SWEEP_WATERMARK_KEY, today_iso)
+        if not claimed:
+            LOGGER.info(
+                "NestQuest missed sweep for %s lost the watermark claim; "
+                "announcing nothing",
+                today_iso,
+            )
+            return []
     LOGGER.info(
         "NestQuest missed sweep for %s built %d event(s)",
         today_iso,
         len(events),
     )
     return events
+