diff --git a/custom_components/nestquest/dao_instances.py b/custom_components/nestquest/dao_instances.py index 93437f1..36d7000 100644 --- a/custom_components/nestquest/dao_instances.py +++ b/custom_components/nestquest/dao_instances.py @@ -277,6 +277,112 @@ class QuestInstancesDao: assert instance is not None return instance + async def upsert_if_valid( + self, + definition_id: int, + child_id: int, + due_date: str, + generated_at: str, + *, + window: str, + ) -> QuestInstanceRecord | None: + """Insert one dated instance iff its tuple is still materializable. + + This is the materialization walk's single write path. Unlike the + generic :meth:`upsert`, which validates only existence and + assignment, this method re-validates EVERY materialization + precondition in the SAME connection-lock-held transaction as the + INSERT — closing the check-then-insert gap a concurrent config + write could otherwise slip through: + + - the definition exists AND is active (a definition deactivated + after the input snapshot is skipped, never written); + - the child exists AND is active (a deactivated child is skipped); + - the (definition, child) assignment link still exists (a removed + assignment is skipped); + - the window is still declared on the definition (a removed window + is skipped), and its CURRENT ``due_time`` is read from the live + row so a due-time edit landing after the snapshot is honored, + never the stale snapshot value. + + Returns the stored instance, or ``None`` when a precondition + failed — the tuple no longer materializable, which the walk SKIPS + rather than raising, so the rest of the batch is unaffected. The + no-past rule and the immutable-completed-instance refusal are + preserved from :meth:`upsert` (those are not configuration + invalidations and still raise ValueError). + """ + _validate_date(due_date, "due_date") + _validate_window(window) + # Fail fast on obviously-past dates before queuing on the lock; + # the authoritative check re-runs under the lock (see upsert). + today = datetime.date.today().isoformat() + if due_date < today: + raise ValueError( + f"instances are never generated in the past: due_date " + f"{due_date!r} is before today {today!r}" + ) + async with _connection_lock(self._database): + async with self._database.transaction(): + today = datetime.date.today().isoformat() + if due_date < today: + raise ValueError( + f"instances are never generated in the past: " + f"due_date {due_date!r} is before today {today!r}" + ) + valid = await self._database.fetch_one( + "SELECT w.due_time 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), + ) + if valid is None: + # A config change invalidated the tuple since the + # snapshot: skip it rather than abort the batch. + return None + due_time = valid[0] + existing = await self._database.fetch_one( + "SELECT 1 FROM completion_events " + "WHERE instance_id = (SELECT id FROM quest_instances " + "WHERE definition_id = ? AND child_id = ? " + "AND due_date = ? AND window = ?)", + (definition_id, child_id, due_date, window), + ) + if existing is not None: + raise ValueError( + f"instance for (definition {definition_id}, " + f"child {child_id}, due_date {due_date!r}, " + f"window {window!r}) already has a completion " + "event and is immutable; it cannot be regenerated" + ) + await self._database.execute( + "INSERT INTO quest_instances (definition_id, child_id, " + "window, due_date, due_time, generated_at) " + "VALUES (?, ?, ?, ?, ?, ?) " + "ON CONFLICT (definition_id, child_id, due_date, " + "window) DO UPDATE SET " + "due_time = excluded.due_time, " + "generated_at = excluded.generated_at", + ( + definition_id, + child_id, + window, + due_date, + due_time, + generated_at, + ), + ) + instance = await self.get( + definition_id, child_id, due_date, window + ) + assert instance is not None + return instance + async def get( self, definition_id: int, diff --git a/custom_components/nestquest/dao_rules.py b/custom_components/nestquest/dao_rules.py index f0da787..ec3f6d9 100644 --- a/custom_components/nestquest/dao_rules.py +++ b/custom_components/nestquest/dao_rules.py @@ -1190,37 +1190,6 @@ 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. diff --git a/custom_components/nestquest/materialize.py b/custom_components/nestquest/materialize.py index 131f569..387380a 100644 --- a/custom_components/nestquest/materialize.py +++ b/custom_components/nestquest/materialize.py @@ -19,16 +19,18 @@ The walk is deterministic and idempotent: - 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 - upserted through :class:`~.dao_instances.QuestInstancesDao.upsert` with - the window's ``due_time`` snapshotted and a ``generated_at`` stamp - shared by the whole batch (one coherent time per run). + written through :class:`~.dao_instances.QuestInstancesDao.upsert_if_valid` + with 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. +tuple's write (an assignment removed, a definition or child deactivated, +a window removed) makes that tuple SKIP rather than abort the batch: +:meth:`~.dao_instances.QuestInstancesDao.upsert_if_valid` re-validates +every materialization precondition (active definition, active child, +live assignment, declared window, and its CURRENT ``due_time``) inside +the SAME transaction as the INSERT, so the check and the write cannot be +split by a racing config edit. The upsert is idempotent on (definition_id, child_id, due_date, window), so re-running the materialization over the same range never duplicates a @@ -40,7 +42,6 @@ import datetime from .dao_instances import QuestInstancesDao from .dao_rules import ( - QuestDefinitionsDao, _validate_date, load_materialization_input, schedule_rule_from_storage, @@ -96,22 +97,6 @@ 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, @@ -125,10 +110,12 @@ async def materialize( of instances upserted for the range (idempotent — a re-run refreshes the same rows, never duplicates them). - 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. + All inputs are read in ONE locked transaction; each tuple's write then + goes through the atomic :meth:`~.dao_instances.QuestInstancesDao.upsert_if_valid`, + which re-validates the tuple and its current ``due_time`` in the same + transaction as the INSERT, 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") @@ -151,7 +138,6 @@ 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() @@ -167,23 +153,14 @@ async def materialize( if not engine.is_present(child.id, cursor): continue for window in windows: - 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 + written = await instances.upsert_if_valid( + definition_id, + child.id, + iso, + generated_at, + window=window.window, + ) + if written is not None: + 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 b6c2736..ff9dace 100644 --- a/tests/test_materialize.py +++ b/tests/test_materialize.py @@ -230,31 +230,36 @@ def test_materialize_is_idempotent(tmp_path) -> None: async def _materialize_with_change( database, start_iso, end_iso, change ) -> int: - """Run materialize, pause before its first per-tuple re-check, apply + """Run materialize, pause before its first atomic write, 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. + BEFORE the first :meth:`QuestInstancesDao.upsert_if_valid` call + acquires the connection lock, so ``change`` commits against the live + state and the write's own in-transaction precondition check then + skips the tuples it invalidated (or, for a presence-only change, + proceeds with the snapped presence). This deterministically + exercises the snapshot-to-insert gap that a config edit can + interleave into. """ - original_check = QuestDefinitionsDao.still_materializable + original_upsert = QuestInstancesDao.upsert_if_valid entered = asyncio.Event() release = asyncio.Event() seen = {"n": 0} - async def _gated_check(self, definition_id, child_id, window): + async def _gated(self, definition_id, child_id, due_date, generated_at, + *, 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) + return await original_upsert( + self, definition_id, child_id, due_date, generated_at, + window=window, + ) - QuestDefinitionsDao.still_materializable = _gated_check + QuestInstancesDao.upsert_if_valid = _gated try: task = asyncio.ensure_future( materialize(database, start_iso, end_iso) @@ -264,7 +269,7 @@ async def _materialize_with_change( release.set() return await task finally: - QuestDefinitionsDao.still_materializable = original_check + QuestInstancesDao.upsert_if_valid = original_upsert def test_materialize_skips_removed_assignee_mid_walk(tmp_path) -> None: