From 3de9f4b970c49e131cddafcd6b3600626e1f5a66 Mon Sep 17 00:00:00 2001 From: orchestrator-agent Date: Sat, 19 Sep 2026 14:15:21 -0400 Subject: [PATCH 1/2] Add the nightly missed-quest sweep Fires one nestquest_quest_missed event per past-due open instance at the configured rollover time and at startup, watermark-guarded in the new nestquest_meta_state table (migration 7, strictly additive). Read-only over the domain tables; never reports an instance that was ever completed. --- custom_components/nestquest/__init__.py | 22 +- custom_components/nestquest/dao_instances.py | 37 +++ custom_components/nestquest/dao_meta.py | 44 ++++ custom_components/nestquest/migrations.py | 11 +- custom_components/nestquest/schema.py | 16 ++ custom_components/nestquest/sweep.py | 103 ++++++++ tests/test_migrations.py | 5 +- tests/test_schema.py | 4 +- tests/test_sweep.py | 239 +++++++++++++++++++ 9 files changed, 473 insertions(+), 8 deletions(-) create mode 100644 custom_components/nestquest/dao_meta.py create mode 100644 custom_components/nestquest/sweep.py create mode 100644 tests/test_sweep.py diff --git a/custom_components/nestquest/__init__.py b/custom_components/nestquest/__init__.py index 5c66d8c..143e9e0 100644 --- a/custom_components/nestquest/__init__.py +++ b/custom_components/nestquest/__init__.py @@ -35,6 +35,7 @@ from .migrations import apply_migrations from .services import async_deregister_services, async_register_services from .store import async_get_db_path from .admin_allowlist import seed_setup_admin +from .sweep import run_missed_sweep _LOGGER = LOGGER @@ -327,18 +328,23 @@ def _register_day_rollover_listener( database: NestQuestDatabase, options: dict[str, Any], ) -> Callable[[], Any]: - """Register the daily materialization listener and return its remover. + """Register the daily rollover listener and return its remover. - Fires once per day at the configured local ``day_rollover_time`` and - materializes the rolling horizon ``[today, today + horizon_days]`` - through the shared :func:`_run_horizon_materialization` path, using the - entry's configured (validated) ``horizon_days``. + Fires once per day at the configured local ``day_rollover_time``: + first re-materializes the rolling horizon + ``[today, today + horizon_days]`` through the shared + :func:`_run_horizon_materialization` path (the entry's configured, + validated ``horizon_days``), then runs the Feature 11 missed sweep + — announcing yesterday's still-open quests as missed AFTER today's + instances exist, watermark-guarded so a re-fire of the listener the + same night announces nothing twice. """ hour, minute = _day_rollover_hour_minute(options) horizon_days = _configured_horizon_days(options) async def _run_materialization(_now: datetime.datetime) -> None: await _run_horizon_materialization(hass, database, horizon_days) + await run_missed_sweep(hass, database) return async_track_time_change( hass, _run_materialization, hour=hour, minute=minute @@ -498,6 +504,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: await _run_horizon_materialization( hass, database, _configured_horizon_days(dict(entry.options)) ) + # The missed sweep runs at startup too (Feature 11): if HA was + # down at the rollover time, the previous night's still-open + # quests are announced as missed now instead of waiting for + # the next midnight. Watermark-guarded, so this is a no-op + # when the night's sweep already ran. + await run_missed_sweep(hass, database) _register_services(hass) # The shared Feature 10 coordinator: one refresh cycle every # entity reads from (CONF_UPDATE_INTERVAL seconds, default diff --git a/custom_components/nestquest/dao_instances.py b/custom_components/nestquest/dao_instances.py index e483d6c..3d30b61 100644 --- a/custom_components/nestquest/dao_instances.py +++ b/custom_components/nestquest/dao_instances.py @@ -493,6 +493,43 @@ class QuestInstancesDao: ) return [_instance_from_row(row) for row in rows] + async def list_missed_unswept( + self, today_iso: str, watermark: str | None + ) -> list[QuestInstanceRecord]: + """Return past-due open instances the sweep has not announced. + + Feature 11's missed-sweep query: ``due_date`` strictly before + ``today_iso`` (past due), at or after ``watermark`` when one is + recorded, and with NO completion event ever (the Feature 08 + append-only contract: an instance that was ever completed is + never reported missed, even when later reversed — history keeps + both rows and the sweep never mutates any). The watermark is + the HA-local date the sweep last ran, and the window is + ``[watermark, today)``: the nightly run entering day D+1 sweeps + exactly day D's instances, so every day's quests are announced + once — at the midnight after their due date — and a same-night + rerun (watermark == today) sees an empty window. Read-only; + ordered deterministically by (due_date, id). + """ + _validate_date(today_iso, "today_iso") + if watermark is not None: + _validate_date(watermark, "watermark") + sql = ( + f"SELECT {_INSTANCE_COLUMNS} FROM quest_instances " + "WHERE due_date < ? " + "AND NOT EXISTS (" + " SELECT 1 FROM completion_events " + " WHERE instance_id = quest_instances.id" + ")" + ) + parameters: tuple = (today_iso,) + if watermark is not None: + sql += " AND due_date >= ?" + parameters += (watermark,) + sql += " ORDER BY due_date, id" + rows = await self._database.fetch_all(sql, parameters) + return [_instance_from_row(row) for row in rows] + async def delete_future_uncompleted( self, definition_id: int, cutoff_date: str, *, today: datetime.date | None = None ) -> int: diff --git a/custom_components/nestquest/dao_meta.py b/custom_components/nestquest/dao_meta.py new file mode 100644 index 0000000..ada8c09 --- /dev/null +++ b/custom_components/nestquest/dao_meta.py @@ -0,0 +1,44 @@ +"""Typed DAO for the integration-owned ``nestquest_meta_state`` table. + +Generic persistent key/value bookkeeping that belongs to the +integration itself rather than any domain table — currently the +Feature 11 missed-sweep watermark. Values are plain strings; callers +interpret them. All SQL for this table lives here per the feature +guardrails, and every method runs through +:class:`~.db.NestQuestDatabase` so nothing blocks the event loop. +""" +from __future__ import annotations + +from .dao_children import _connection_lock +from .db import NestQuestDatabase + +_TABLE = "nestquest_meta_state" + + +class MetaStateDao: + """Typed async access to the ``nestquest_meta_state`` table.""" + + def __init__(self, database: NestQuestDatabase) -> None: + self._database = database + + async def get(self, key: str) -> str | None: + """Return the stored value for ``key``, or None when absent.""" + row = await self._database.fetch_one( + f"SELECT value FROM {_TABLE} WHERE key = ?", (key,) + ) + return row[0] if row is not None else None + + async def set(self, key: str, value: str) -> None: + """Upsert one key/value row atomically. + + The write runs inside one transaction under the shared + connection lock, so a racing reader either sees the old value + or the fully-written new one, never a half-applied state. + """ + async with _connection_lock(self._database): + async with self._database.transaction(): + await self._database.execute( + f"INSERT INTO {_TABLE} (key, value) VALUES (?, ?) " + "ON CONFLICT (key) DO UPDATE SET value = excluded.value", + (key, value), + ) \ No newline at end of file diff --git a/custom_components/nestquest/migrations.py b/custom_components/nestquest/migrations.py index 2040366..cf40a7a 100644 --- a/custom_components/nestquest/migrations.py +++ b/custom_components/nestquest/migrations.py @@ -28,7 +28,10 @@ copying each definition's assignee across. Migration 4 adds the windows table, migration 5 rebuilds ``quest_instances`` onto the widened (definition_id, child_id, due_date, window) key, and migration 6 rebuilds ``completion_events`` with the D-008 ``actor_child_id`` -column. Each callable migration runs inside its own transaction +column. Migration 7 adds the generic ``nestquest_meta_state`` +key/value table for integration-owned bookkeeping (Feature 11's +missed-sweep watermark). Each callable migration runs inside its own +transaction together with its version stamp, so a crash mid-step rolls the statements and the stamp back together. @@ -80,6 +83,7 @@ from .schema import ( SCHEMA_V1_STATEMENTS, SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL, SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL, + SCHEMA_V7_META_STATE_DDL, ) LOGGER = logging.getLogger(__name__) @@ -414,6 +418,11 @@ MIGRATIONS: Sequence[MigrationStep] = [ # actor_child_id column and its actor-pair coherence CHECK, # backfilling legacy panel rows to their own child_id. _add_actor_child_id, + # Migration 7 (Feature 11): the generic meta_state key/value table + # — a plain new table, so CREATE IF NOT EXISTS suffices for both + # fresh databases (which already made it in migration 1) and + # pre-v7 databases. Strictly additive; no existing table changes. + list(SCHEMA_V7_META_STATE_DDL), ] diff --git a/custom_components/nestquest/schema.py b/custom_components/nestquest/schema.py index ab1fb35..b3c3223 100644 --- a/custom_components/nestquest/schema.py +++ b/custom_components/nestquest/schema.py @@ -384,6 +384,21 @@ SCHEMA_V1_COMPLETION_EVENTS_DDL: list[str] = [ """, ] +#: Generic integration-owned key/value state (schema version 7, +#: Feature 11): persistent bookkeeping that belongs to the integration +#: rather than any domain table — currently the missed-sweep +#: watermark. ``value`` is a plain string; callers interpret it. +META_STATE_TABLE_SQL = """( + key TEXT NOT NULL PRIMARY KEY, + value TEXT NOT NULL +)""" + +SCHEMA_V7_META_STATE_DDL: list[str] = [ + f""" + CREATE TABLE IF NOT EXISTS nestquest_meta_state {META_STATE_TABLE_SQL} + """, +] + SCHEMA_V1_STATEMENTS: list[str] = [ *SCHEMA_V1_CHILDREN_DDL, *SCHEMA_V1_ADMIN_USERS_DDL, @@ -395,6 +410,7 @@ SCHEMA_V1_STATEMENTS: list[str] = [ *SCHEMA_V1_PRESENCE_OVERRIDES_DDL, *SCHEMA_V1_QUEST_INSTANCES_DDL, *SCHEMA_V1_COMPLETION_EVENTS_DDL, + *SCHEMA_V7_META_STATE_DDL, ] diff --git a/custom_components/nestquest/sweep.py b/custom_components/nestquest/sweep.py new file mode 100644 index 0000000..a36cdf4 --- /dev/null +++ b/custom_components/nestquest/sweep.py @@ -0,0 +1,103 @@ +"""The nightly missed-quest sweep (Feature 11). + +At the configured day-rollover time (the same HA-local listener that +re-materializes the rolling horizon) this job finds every instance +whose ``due_date`` is before the HA-local date, that has NO completion +event, and that falls after the persisted sweep watermark — and fires +ONE ``nestquest_quest_missed`` event per instance with the documented +payload. The sweep is read-only over the domain tables: it never +mutates an instance and never writes a completion event (``missed`` +is derived state, never an event_type — Feature 08's contract). + +Idempotency is the watermark: after a run, ``nestquest_meta_state`` +records the HA-local date the sweep ran, and the next run only +sweeps instances due in ``[watermark, today)``. A second run the +same night therefore fires nothing new, and a run after days of +downtime sweeps exactly the accumulated window once — no +per-instance journal, no re-announcing old misses. An instance swept +as missed that is 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. +""" +from __future__ import annotations + +import datetime +import logging +from zoneinfo import ZoneInfo + +from .const import EVENT_QUEST_MISSED +from .dao_children import ChildrenDao +from .dao_instances import QuestInstancesDao +from .dao_meta import MetaStateDao +from .dao_rules import QuestDefinitionsDao +from .db import NestQuestDatabase + +LOGGER = logging.getLogger(__name__) + +#: meta_state key holding the HA-local date the sweep last ran. +SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date" + +#: UTC ISO-8601 timestamp policy shared with the DAO layers. +_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00" + + +def _now_stamp() -> str: + return datetime.datetime.now(datetime.timezone.utc).strftime( + _UTC_TIMESTAMP_FORMAT + ) + + +async def run_missed_sweep( + hass, + database: NestQuestDatabase, + *, + today: datetime.date | None = None, +) -> int: + """Fire one missed event per unswept past-due open instance. + + ``today`` optionally pins the HA-local date (threaded like the + materialization and completion layers); when omitted it is read + 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). + """ + today_date = ( + today + if today is not None + else datetime.datetime.now(ZoneInfo(hass.config.time_zone)).date() + ) + 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 + + 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 + ) + return fired \ No newline at end of file diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 7cf0624..89195e5 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -10,6 +10,7 @@ from unittest.mock import AsyncMock import pytest from custom_components.nestquest.db import NestQuestDatabase +from custom_components.nestquest.schema import SCHEMA_V7_META_STATE_DDL from custom_components.nestquest.migrations import ( MIGRATIONS, VERSION_TABLE, @@ -110,6 +111,7 @@ def test_fresh_file_migration_creates_all_v1_tables(tmp_path) -> None: "presence_overrides", "quest_instances", "completion_events", + "nestquest_meta_state", VERSION_TABLE, } assert _tables(database) == expected @@ -159,7 +161,8 @@ def test_migration_list_shape() -> None: assert MIGRATIONS[3] == SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL assert callable(MIGRATIONS[4]) assert callable(MIGRATIONS[5]) - assert len(MIGRATIONS) == 6 + assert MIGRATIONS[6] == SCHEMA_V7_META_STATE_DDL + assert len(MIGRATIONS) == 7 # --------------------------------------------------------------------------- diff --git a/tests/test_schema.py b/tests/test_schema.py index f869176..50a26a2 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -11,6 +11,7 @@ from custom_components.nestquest.schema import ( SCHEMA_V1_ADMIN_USERS_DDL, SCHEMA_V1_CHILDREN_DDL, SCHEMA_V1_COMPLETION_EVENTS_DDL, + SCHEMA_V7_META_STATE_DDL, SCHEMA_V1_PRESENCE_OVERRIDES_DDL, SCHEMA_V1_PRESENCE_SCHEDULES_DDL, SCHEMA_V1_SCHEDULE_RULES_DDL, @@ -168,7 +169,7 @@ def _insert_override(database, child_id, **overrides): # --------------------------------------------------------------------------- -def test_schema_v1_statements_compose_the_eight_tables() -> None: +def test_schema_v1_statements_compose_the_nine_tables() -> None: assert SCHEMA_V1_STATEMENTS == [ *SCHEMA_V1_CHILDREN_DDL, *SCHEMA_V1_ADMIN_USERS_DDL, @@ -180,6 +181,7 @@ def test_schema_v1_statements_compose_the_eight_tables() -> None: *SCHEMA_V1_PRESENCE_OVERRIDES_DDL, *SCHEMA_V1_QUEST_INSTANCES_DDL, *SCHEMA_V1_COMPLETION_EVENTS_DDL, + *SCHEMA_V7_META_STATE_DDL, ] diff --git a/tests/test_sweep.py b/tests/test_sweep.py new file mode 100644 index 0000000..17ee530 --- /dev/null +++ b/tests/test_sweep.py @@ -0,0 +1,239 @@ +"""Tests for the nightly missed-quest sweep (Feature 11, task 1).""" +from __future__ import annotations + +import datetime +import re + +from conftest import wire_entry_to_registry + +from custom_components.nestquest import async_setup_entry +from custom_components.nestquest.children import create_child, list_children +from custom_components.nestquest.completion import ( + complete_instance, + uncomplete_instance, +) +from custom_components.nestquest.const import EVENT_QUEST_MISSED +from custom_components.nestquest.dao_instances import ( + CompletionEventsDao, + QuestInstancesDao, +) +from custom_components.nestquest.dao_meta import MetaStateDao +from custom_components.nestquest.materialize import materialize +from custom_components.nestquest.quest_definitions import ( + create_quest_definition, +) +from custom_components.nestquest.recurrence import ScheduleRule +from custom_components.nestquest.sweep import ( + SWEEP_WATERMARK_KEY, + run_missed_sweep, +) + +_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00$") + + +async def _seed(hass, make_entry) -> tuple: + entry = wire_entry_to_registry(make_entry(), hass.registry) + assert await async_setup_entry(hass, entry) is True + database = entry.runtime_data.database + ada = await create_child(database, "Ada") + bo = await create_child(database, "Bo") + today = datetime.date.today() + await create_quest_definition( + database, + "Brush teeth", + ScheduleRule.from_dict( + {"rule_type": "daily", "start_date": today.isoformat()} + ), + [ada.id, bo.id], + ["morning"], + ) + end = (today + datetime.timedelta(days=3)).isoformat() + await materialize(database, today.isoformat(), end, today=today) + return entry, ada, bo + + +async def test_sweep_fires_one_event_per_past_due_open_instance( + hass, make_entry +) -> None: + entry, ada, bo = await _seed(hass, make_entry) + database = entry.runtime_data.database + today = datetime.date.today() + tomorrow = today + datetime.timedelta(days=1) + + fired = await run_missed_sweep(hass, database, today=tomorrow) + assert fired == 2, "one per child for today's now-past-due instance" + + events = hass.bus.fired(EVENT_QUEST_MISSED) + assert len(events) == 2 + payload = events[0] + assert payload["child_name"] in ("Ada", "Bo") + assert payload["quest_title"] == "Brush teeth" + assert payload["window"] == "morning" + assert payload["due_date"] == today.isoformat() + assert payload["due_time"] is None + assert _TIMESTAMP.match(payload["occurred_at"]) + assert {payload["child_id"] for payload in events} == {ada.id, bo.id} + # The sweep is read-only over the domain tables. + assert await _count_events(database) == 0 + + +async def test_sweep_is_idempotent_same_night(hass, make_entry) -> None: + entry, _ada, _bo = await _seed(hass, make_entry) + database = entry.runtime_data.database + tomorrow = datetime.date.today() + datetime.timedelta(days=1) + + first = await run_missed_sweep(hass, database, today=tomorrow) + second = await run_missed_sweep(hass, database, today=tomorrow) + assert first == 2 + assert second == 0, "the watermark makes a rerun fire nothing" + assert len(hass.bus.fired(EVENT_QUEST_MISSED)) == 2 + assert await MetaStateDao(database).get(SWEEP_WATERMARK_KEY) == ( + tomorrow.isoformat() + ) + + +async def test_sweep_only_fires_new_instances_after_watermark( + hass, make_entry +) -> None: + entry, ada, _bo = await _seed(hass, make_entry) + database = entry.runtime_data.database + today = datetime.date.today() + tomorrow = today + datetime.timedelta(days=1) + day_after = today + datetime.timedelta(days=2) + + # First sweep one day ahead: today's instances fire. + fired = await run_missed_sweep(hass, database, today=tomorrow) + assert fired == 2 + first_events = hass.bus.fired(EVENT_QUEST_MISSED) + assert all( + payload["due_date"] == today.isoformat() + for payload in first_events + ) + + # One more day passes: only tomorrow's instances fire now. + fired = await run_missed_sweep(hass, database, today=day_after) + assert fired == 2 + all_events = hass.bus.fired(EVENT_QUEST_MISSED) + assert len(all_events) == 4 + assert all( + payload["due_date"] == tomorrow.isoformat() + for payload in all_events[2:] + ) + + +async def test_sweep_never_reports_completed_or_reversed_instances( + hass, make_entry +) -> None: + entry, ada, bo = await _seed(hass, make_entry) + database = entry.runtime_data.database + today = datetime.date.today() + tomorrow = today + datetime.timedelta(days=1) + + instances_ada = await QuestInstancesDao(database).list_by_date_range( + ada.id, today.isoformat(), today.isoformat() + ) + instances_bo = await QuestInstancesDao(database).list_by_date_range( + bo.id, today.isoformat(), today.isoformat() + ) + # Ada completed hers; Bo completed his then reversed it. + await complete_instance( + database, + instances_ada[0].id, + actor_source="panel", + actor_child_id=ada.id, + ) + await complete_instance( + database, + instances_bo[0].id, + actor_source="panel", + actor_child_id=bo.id, + ) + await uncomplete_instance( + database, + instances_bo[0].id, + actor_source="user", + actor_user_id="admin-1", + ) + + fired = await run_missed_sweep(hass, database, today=tomorrow) + assert fired == 0, ( + "no-completion-event rule excludes both the completed and the " + "reversed instance (append-only history keeps both)" + ) + assert hass.bus.fired(EVENT_QUEST_MISSED) == [] + + +async def test_sweep_skips_instance_completed_after_sweep( + hass, make_entry +) -> None: + """A missed instance completed later is never re-announced; the + late completion still lands in the append-only log.""" + entry, ada, _bo = await _seed(hass, make_entry) + database = entry.runtime_data.database + today = datetime.date.today() + tomorrow = today + datetime.timedelta(days=1) + day3 = today + datetime.timedelta(days=2) + + await run_missed_sweep(hass, database, today=tomorrow) + instance = ( + await QuestInstancesDao(database).list_by_date_range( + ada.id, today.isoformat(), today.isoformat() + ) + )[0] + await complete_instance( + database, + instance.id, + actor_source="panel", + actor_child_id=ada.id, + ) + fired = await run_missed_sweep(hass, database, today=day3) + assert fired == 0 or all( + payload["instance_id"] != instance.id + for payload in hass.bus.fired(EVENT_QUEST_MISSED)[2:] + ) + + +async def test_rollover_listener_runs_sweep_after_materialization( + hass, make_entry, monkeypatch +) -> None: + import custom_components.nestquest as nestquest + + calls = [] + + async def _record_sweep(hass_arg, database, **kwargs): + calls.append("sweep") + + monkeypatch.setattr(nestquest, "run_missed_sweep", _record_sweep) + + entry = wire_entry_to_registry( + make_entry(options={"day_rollover_time": "00:00"}), hass.registry + ) + assert await async_setup_entry(hass, entry) is True + assert calls == ["sweep"], "startup runs the sweep once" + + await hass.time_change.fire() + assert calls == ["sweep", "sweep"], "the rollover listener sweeps too" + + +async def test_v6_database_migrates_to_v7_meta_state( + hass, make_entry +) -> None: + """A database stamped at version 6 gains the meta_state table and + version 7 on the next startup; existing rows are untouched.""" + entry = wire_entry_to_registry(make_entry(), hass.registry) + assert await async_setup_entry(hass, entry) is True + database = entry.runtime_data.database + version = await database.fetch_one( + "SELECT version FROM nestquest_schema_version" + ) + assert version == (7,) + tables = await database.fetch_all( + "SELECT name FROM sqlite_master WHERE type='table' " + "AND name='nestquest_meta_state'" + ) + assert tables, "migration 7 created the meta_state table" + + +async def _count_events(database) -> int: + rows = await database.fetch_all("SELECT COUNT(*) FROM completion_events") + return rows[0][0] From 8cd43141a9cbd74b7d930fd75db087f7be8f5da1 Mon Sep 17 00:00:00 2001 From: orchestrator-agent Date: Sat, 19 Sep 2026 14:16:50 -0400 Subject: [PATCH 2/2] 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 | 85 +++++++++++++++++++--------- tests/test_sweep.py | 20 +++++++ 2 files changed, 77 insertions(+), 28 deletions(-) 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