fix: single coherent input snapshot and skip-invalidated-tuple (NQ-75-001, NQ-75-002)

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)

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"
-            )
+            )
+
+
+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

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