fix: atomic materialization upsert validates active/assignee/window (NQ-75-CR-001)
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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
|
||||
+17
-12
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user