diff --git a/custom_components/nestquest/dao_presence.py b/custom_components/nestquest/dao_presence.py index 07f3111..cf1707a 100644 --- a/custom_components/nestquest/dao_presence.py +++ b/custom_components/nestquest/dao_presence.py @@ -279,7 +279,7 @@ class PresenceOverridesDao: return result.rowcount > 0 -async def load_snapshot( +async def read_snapshot_unlocked( database: NestQuestDatabase, child_ids: list[int], range_start: str, @@ -288,7 +288,13 @@ async def load_snapshot( dict[int, PresenceScheduleRecord], dict[int, list[PresenceOverrideRecord]], ]: - """Read schedules and overrides for ``child_ids`` in ONE transaction. + """Read schedules and overrides for ``child_ids`` WITHOUT locking. + + MUST be called inside the connection lock and an open transaction, so + the schedules and overrides returned here stay coherent with every + other table the caller reads inside that same transaction (the + materialization walk's single input snapshot does exactly that). The + date range comes already validated by the caller. Returns ``(schedules, overrides)``: @@ -299,46 +305,24 @@ async def load_snapshot( ``[range_start, range_end]`` (inclusive boundaries), ordered by start date; a child with no matching override is absent from the mapping. - Both tables are read inside one BEGIN..COMMIT span under the - connection-scoped lock, so the returned pairing is a coherent snapshot - — a concurrent schedule upsert or override create cannot interleave - between the two reads. Children are validated as existing up front so - an unknown ``child_id`` raises ValueError rather than being silently - read as "present every day", mirroring - :meth:`PresenceSchedulesDao.get_by_child`. + ``child_ids`` must name existing children (the caller derives them + from a trusted snapshot, within the same transaction). """ - _validate_date(range_start, "range_start") - _validate_date(range_end, "range_end") - if range_end < range_start: - raise ValueError( - "range_end must be on or after range_start, got " - f"{range_end!r} < {range_start!r}" - ) children = sorted(set(child_ids)) if not children: return {}, {} placeholders = ",".join("?" for _ in children) - async with _connection_lock(database): - async with database.transaction(): - existing_rows = await database.fetch_all( - f"SELECT id FROM children WHERE id IN ({placeholders})", - tuple(children), - ) - existing = {row[0] for row in existing_rows} - unknown = sorted(set(children) - existing) - if unknown: - raise ValueError(f"child {unknown[0]} does not exist") - schedule_rows = await database.fetch_all( - f"SELECT {_SCHEDULE_COLUMNS} FROM presence_schedules " - f"WHERE child_id IN ({placeholders})", - tuple(children), - ) - override_rows = await database.fetch_all( - f"SELECT {_OVERRIDE_COLUMNS} FROM presence_overrides " - f"WHERE child_id IN ({placeholders}) AND end_date >= ? " - "AND start_date <= ? ORDER BY start_date", - tuple(children) + (range_start, range_end), - ) + schedule_rows = await database.fetch_all( + f"SELECT {_SCHEDULE_COLUMNS} FROM presence_schedules " + f"WHERE child_id IN ({placeholders})", + tuple(children), + ) + override_rows = await database.fetch_all( + f"SELECT {_OVERRIDE_COLUMNS} FROM presence_overrides " + f"WHERE child_id IN ({placeholders}) AND end_date >= ? " + "AND start_date <= ? ORDER BY start_date", + tuple(children) + (range_start, range_end), + ) schedules = { record.child_id: record for record in map(_schedule_from_row, schedule_rows) diff --git a/custom_components/nestquest/dao_rules.py b/custom_components/nestquest/dao_rules.py index 5669368..f0da787 100644 --- a/custom_components/nestquest/dao_rules.py +++ b/custom_components/nestquest/dao_rules.py @@ -951,6 +951,21 @@ class QuestDefinitionsDao: definitions = await self.list_active() return await self._snapshots_for(definitions) + async def read_snapshots_active(self) -> list[QuestDefinitionSnapshot]: + """Return every active definition's snapshot WITHOUT locking. + + Mirrors :meth:`list_snapshots_active` minus the lock and + transaction: the definitions and their rules, assignees and + windows are read on the caller's connection under whatever + transaction the caller is already holding, so they stay coherent + with every OTHER table the caller reads inside that same + transaction. MUST be called inside the connection lock and an + open transaction (the single-transaction input snapshot the + materialization walk uses is that caller). + """ + definitions = await self.list_active() + return await self._snapshots_for(definitions) + async def list_snapshots_by_child( self, child_id: int ) -> list[QuestDefinitionSnapshot]: @@ -1175,6 +1190,37 @@ class QuestDefinitionsDao: ) return result.rowcount > 0 + async def still_materializable( + self, definition_id: int, child_id: int, window: str + ) -> bool: + """Return True when the (definition, child, window) tuple is still + a valid materialization target. + + Checks, in one query under the connection lock, that the + definition exists AND is active, the child exists AND is active, + the (definition, child) assignment link is still present, and the + window is still declared on the definition. The materialization + walk re-checks each tuple against this verdict right before its + upsert so a config change landed between the input snapshot and + the upsert (an assignment removed, a definition or child + deactivated, a window removed) causes that tuple to be SKIPPED + instead of aborting the batch. This is a point read under the + lock; the walk additionally treats the upsert's own "does not + exist" / "not assigned" rejections as skippable, closing the + assignment/existence gap. + """ + async with _connection_lock(self._database): + row = await self._database.fetch_one( + "SELECT 1 FROM quest_definitions d " + "JOIN quest_definition_assignees a ON a.definition_id = d.id " + "JOIN children c ON c.id = a.child_id " + "JOIN quest_definition_windows w ON w.definition_id = d.id " + "WHERE d.id = ? AND a.child_id = ? AND w.window = ? " + "AND d.is_active = 1 AND c.is_active = 1 LIMIT 1", + (definition_id, child_id, window), + ) + return row is not None + async def _validate_assignable_child(self, child_id: int) -> None: """Raise ValueError unless the child exists and is active. @@ -1189,4 +1235,47 @@ class QuestDefinitionsDao: if not child[0]: raise ValueError( f"child {child_id} is inactive and cannot be assigned" - ) \ No newline at end of file + ) + + +async def load_materialization_input( + database: NestQuestDatabase, + range_start: str, + range_end: str, +): + """Read every materialization input in ONE locked transaction. + + Returns ``(snapshots, schedules, overrides)``: + + - ``snapshots``: every active definition with its rule, assignees and + windows (the :class:`QuestDefinitionSnapshot` shape). + - ``schedules`` / ``overrides``: the presence schedules and + range-scoped overrides of every assignee child, in + :mod:`.dao_presence`'s record shapes. + + The whole read — definitions, rules, assignees, windows, and the + assignees' presence schedules and overrides — runs inside a single + BEGIN..COMMIT span under the connection lock, so a concurrent edit + (definition, assignment, window, or presence) cannot interleave + between the parts and the walk never sees a mixed-time view. Each + table's SQL stays in its own DAO module: the presence portion is + delegated to :mod:`.dao_presence` (a lazy import here, since that + module imports this one). + """ + definitions_dao = QuestDefinitionsDao(database) + async with _connection_lock(database): + async with database.transaction(): + snapshots = await definitions_dao.read_snapshots_active() + child_ids = sorted( + { + child.id + for snapshot in snapshots + for child in snapshot.assignees + } + ) + from . import dao_presence # lazy: dao_presence imports us + + schedules, overrides = await dao_presence.read_snapshot_unlocked( + database, child_ids, range_start, range_end + ) + return snapshots, schedules, overrides \ No newline at end of file diff --git a/custom_components/nestquest/materialize.py b/custom_components/nestquest/materialize.py index 2a41a2e..131f569 100644 --- a/custom_components/nestquest/materialize.py +++ b/custom_components/nestquest/materialize.py @@ -8,13 +8,14 @@ generation are separate later tasks and live elsewhere. The walk is deterministic and idempotent: -- Every active definition is read with its decoded - :class:`~.recurrence.ScheduleRule`, assignees and windows through the - Feature 06 snapshot helpers, so each bundle is a coherent read. +- All inputs — the active definitions with their rules, assignees and + windows, AND the assignee children's presence schedules and scoped + overrides — are read inside ONE locked transaction (see + :func:`~.dao_rules.load_materialization_input`), so the walk sees a + single coherent snapshot rather than a definition state from one + instant and a presence state from another. - Presence is snapshotted into an immutable - :class:`~.presence.PresenceEngine` from the schedules and overrides of - every assignee child, read in one locked transaction (see - :func:`~.dao_presence.load_snapshot`). + :class:`~.presence.PresenceEngine` from that same coherent read. - For each date in the range, each definition whose decoded rule :func:`~.recurrence.occurs_on` on that date, each assignee who is ``is_present`` on that date and each declared window, one instance is @@ -22,6 +23,13 @@ The walk is deterministic and idempotent: the window's ``due_time`` snapshotted and a ``generated_at`` stamp shared by the whole batch (one coherent time per run). +A config change that lands AFTER the snapshot is read but BEFORE a +tuple's upsert (an assignment removed, a definition or child +deactivated, a window removed) makes that tuple SKIP rather than abort +the batch: each tuple is re-validated against the current state right +before its upsert, and only the narrow "no longer valid" conditions are +treated as skippable. + The upsert is idempotent on (definition_id, child_id, due_date, window), so re-running the materialization over the same range never duplicates a row; skipped-completed handling is deliberately NOT done here. @@ -31,10 +39,10 @@ from __future__ import annotations import datetime from .dao_instances import QuestInstancesDao -from .dao_presence import load_snapshot from .dao_rules import ( QuestDefinitionsDao, _validate_date, + load_materialization_input, schedule_rule_from_storage, schedule_rule_storage_from_record, ) @@ -88,6 +96,22 @@ def _build_engine( return PresenceEngine(schedules, overrides) +def _is_deleted_tuple(error: ValueError) -> bool: + """True for the narrow 'tuple no longer valid' upsert rejections. + + A change landing between the per-tuple validity re-check and the + upsert can still surface as the upsert's "definition does not exist" + or "not assigned to child" error; those are skippable. Every other + rejection (past due date, immutable completed instance, bad window) + is a real error and propagates. + """ + message = str(error) + return ( + "does not exist" in message + or "is not assigned to child" in message + ) + + async def materialize( database: NestQuestDatabase, start_date: str, @@ -98,13 +122,13 @@ async def materialize( ``start_date`` and ``end_date`` must be strict ISO calendar dates (YYYY-MM-DD) with ``end_date`` on or after ``start_date``; anything else raises ValueError naming the offending field. Returns the number - of instances upserted (idempotent — a re-run for the same batch - refreshes the same rows, never duplicates them). + of instances upserted for the range (idempotent — a re-run refreshes + the same rows, never duplicates them). - The active definitions and the presence inputs are each read as a - coherent snapshot, and the presence schedule/overrides are read under - one locked transaction so the walk sees a single stable view of who - is where on each date. + All inputs are read in ONE locked transaction, and each tuple is + re-validated against the current state just before its upsert so a + config change that arrives mid-batch skips the affected tuples rather + than aborting the batch or leaving a partial result. """ _validate_date(start_date, "start_date") _validate_date(end_date, "end_date") @@ -114,17 +138,8 @@ async def materialize( f"{end_date!r} < {start_date!r}" ) - snapshots = await QuestDefinitionsDao(database).list_snapshots_active() - assignee_child_ids = sorted( - { - child.id - for snapshot in snapshots - for child in snapshot.assignees - } - ) - - schedules_records, overrides_records = await load_snapshot( - database, assignee_child_ids, start_date, end_date + snapshots, schedules_records, overrides_records = await ( + load_materialization_input(database, start_date, end_date) ) engine = _build_engine(schedules_records, overrides_records) @@ -136,6 +151,7 @@ async def materialize( generated_at = _now_stamp() instances = QuestInstancesDao(database) + definitions_dao = QuestDefinitionsDao(database) start = datetime.datetime.strptime(start_date, "%Y-%m-%d").date() end = datetime.datetime.strptime(end_date, "%Y-%m-%d").date() @@ -151,14 +167,23 @@ async def materialize( if not engine.is_present(child.id, cursor): continue for window in windows: - await instances.upsert( - definition_id, - child.id, - iso, - generated_at, - window=window.window, - due_time=window.due_time, - ) + if not await definitions_dao.still_materializable( + definition_id, child.id, window.window + ): + continue + try: + await instances.upsert( + definition_id, + child.id, + iso, + generated_at, + window=window.window, + due_time=window.due_time, + ) + except ValueError as error: + if _is_deleted_tuple(error): + continue + raise count += 1 cursor += datetime.timedelta(days=1) return count \ No newline at end of file diff --git a/tests/test_materialize.py b/tests/test_materialize.py index 1e6f621..b6c2736 100644 --- a/tests/test_materialize.py +++ b/tests/test_materialize.py @@ -9,6 +9,7 @@ import pytest from custom_components.nestquest.dao_children import ChildrenDao from custom_components.nestquest.dao_instances import QuestInstancesDao from custom_components.nestquest.dao_presence import PresenceSchedulesDao +from custom_components.nestquest.dao_rules import QuestDefinitionsDao from custom_components.nestquest.db import NestQuestDatabase from custom_components.nestquest.materialize import materialize from custom_components.nestquest.migrations import apply_migrations @@ -209,11 +210,243 @@ def test_materialize_is_idempotent(tmp_path) -> None: first = await materialize(database, start_iso, end_iso) second = await materialize(database, start_iso, end_iso) assert first == second == 7 + # Running twice refreshes the SAME seven rows: exactly one + # physical row per (definition_id, child_id, due_date, window) + # key, never a duplicate. + records = await QuestInstancesDao(database).list_by_date_range( + child.id, start_iso, end_iso + ) + keys = [ + (r.definition_id, r.child_id, r.due_date, r.window) + for r in records + ] + assert len(keys) == 7 + assert len(set(keys)) == 7 return first _with_db(tmp_path, "materialize-idempotent.db")(_body) +async def _materialize_with_change( + database, start_iso, end_iso, change +) -> int: + """Run materialize, pause before its first per-tuple re-check, apply + ``change``, then release. Returns the materialize count. + + The gate parks the walk AFTER the single input snapshot is read but + BEFORE the first :meth:`QuestDefinitionsDao.still_materializable` + re-check acquires the connection lock, so ``change`` commits against + the live state and the walk's per-tuple re-check then skips the + tuples it invalidated (or, for a presence-only change, continues with + the snapped presence). This deterministically exercises the + snapshot-to-upsert gap that a config edit can interleave into. + """ + original_check = QuestDefinitionsDao.still_materializable + entered = asyncio.Event() + release = asyncio.Event() + seen = {"n": 0} + + async def _gated_check(self, definition_id, child_id, window): + seen["n"] += 1 + if seen["n"] == 1: + seen["first"] = (definition_id, child_id, window) + entered.set() + await release.wait() + return await original_check(self, definition_id, child_id, window) + + QuestDefinitionsDao.still_materializable = _gated_check + try: + task = asyncio.ensure_future( + materialize(database, start_iso, end_iso) + ) + await entered.wait() + await change() + release.set() + return await task + finally: + QuestDefinitionsDao.still_materializable = original_check + + +def test_materialize_skips_removed_assignee_mid_walk(tmp_path) -> None: + async def _body(database): + children = ChildrenDao(database) + a = await children.create("Ada", NOW) + c = await children.create("Cleo", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=2)).isoformat() + created = await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [a.id, c.id], + ["morning"], + ) + + async def _remove_assignee(): + await QuestDefinitionsDao(database).remove_assignee( + created.definition.id, a.id + ) + + count = await _materialize_with_change( + database, start_iso, end_iso, _remove_assignee + ) + assert count == 3 # Cleo's three days survive + assert await QuestInstancesDao(database).list_by_date_range( + a.id, start_iso, end_iso + ) == [] + assert len(await QuestInstancesDao(database).list_by_date_range( + c.id, start_iso, end_iso + )) == 3 + return count + + _with_db(tmp_path, "materialize-removed-assignee.db")(_body) + + +def test_materialize_skips_deactivated_definition_mid_walk(tmp_path) -> None: + async def _body(database): + children = ChildrenDao(database) + a = await children.create("Ada", NOW) + c = await children.create("Cleo", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=2)).isoformat() + created = await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [a.id, c.id], + ["morning"], + ) + + async def _deactivate(): + await QuestDefinitionsDao(database).set_active( + created.definition.id, False + ) + + count = await _materialize_with_change( + database, start_iso, end_iso, _deactivate + ) + assert count == 0 + assert await QuestInstancesDao(database).list_by_date_range( + a.id, start_iso, end_iso + ) == [] + assert await QuestInstancesDao(database).list_by_date_range( + c.id, start_iso, end_iso + ) == [] + return count + + _with_db(tmp_path, "materialize-deactivated-definition.db")(_body) + + +def test_materialize_skips_removed_window_mid_walk(tmp_path) -> None: + async def _body(database): + children = ChildrenDao(database) + a = await children.create("Ada", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=2)).isoformat() + created = await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [a.id], + [("morning", "09:00")], + ) + + async def _remove_window(): + await QuestDefinitionsDao(database).remove_window( + created.definition.id, "morning" + ) + + count = await _materialize_with_change( + database, start_iso, end_iso, _remove_window + ) + assert count == 0 + assert await QuestInstancesDao(database).list_by_date_range( + a.id, start_iso, end_iso + ) == [] + return count + + _with_db(tmp_path, "materialize-removed-window.db")(_body) + + +def test_materialize_skips_deactivated_child_mid_walk(tmp_path) -> None: + async def _body(database): + children = ChildrenDao(database) + a = await children.create("Ada", NOW) + c = await children.create("Cleo", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=2)).isoformat() + created = await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [a.id, c.id], + ["morning"], + ) + + async def _deactivate_child(): + await ChildrenDao(database).set_active(a.id, False) + + count = await _materialize_with_change( + database, start_iso, end_iso, _deactivate_child + ) + assert count == 3 # Cleo's three days survive + assert await QuestInstancesDao(database).list_by_date_range( + a.id, start_iso, end_iso + ) == [] + assert len(await QuestInstancesDao(database).list_by_date_range( + c.id, start_iso, end_iso + )) == 3 + return count + + _with_db(tmp_path, "materialize-deactivated-child.db")(_body) + + +def test_materialize_uses_single_presence_snapshot_mid_walk(tmp_path) -> None: + async def _body(database): + from custom_components.nestquest.dao_presence import ( + PresenceOverridesDao, + ) + + children = ChildrenDao(database) + overrides = PresenceOverridesDao(database) + child = await children.create("Ada", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=2)).isoformat() + await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [child.id], + ["morning"], + ) + + async def _add_blocking_override(): + await overrides.create( + child.id, start_iso, end_iso, False, note="away" + ) + + # Presence is read once, inside the input snapshot; a blocking + # override landing AFTER that read must not make the walk panic + # or half-apply it. The walk stays fully governed by the + # snapshot it read: Ada was present every day there, so all + # three days still materialize, deterministically. + count = await _materialize_with_change( + database, start_iso, end_iso, _add_blocking_override + ) + assert count == 3 + assert len(await QuestInstancesDao(database).list_by_date_range( + child.id, start_iso, end_iso + )) == 3 + return count + + _with_db(tmp_path, "materialize-presence-snapshot.db")(_body) + + def test_materialize_honours_overrides(tmp_path) -> None: async def _body(database): from custom_components.nestquest.dao_presence import (