fix: atomic complete_instance check+append

custom_components/nestquest/completion.py

@@ -25,9 +25,10 @@ is no permission check here (that is Feature 09).
 """
 from __future__ import annotations
 
+import asyncio
 import datetime
 
-from .dao_children import _connection_lock
+from .dao_children import _CONNECTION_LOCKS, _connection_lock
 from .dao_instances import (
     EVENT_COMPLETED,
     EVENT_UNCOMPLETED,
@@ -62,6 +63,63 @@ def _stamp_occurred_at(now: datetime.datetime | None) -> str:
     )
 
 
+class _TaskReentrantLock:
+    """Shared connection lock the owning task may re-enter."""
+
+    def __init__(self, inner: asyncio.Lock) -> None:
+        self._inner = inner
+        self._task: asyncio.Task | None = None
+        self._depth = 0
+
+    def locked(self) -> bool:
+        return self._inner.locked()
+
+    async def acquire(self) -> bool:
+        task = asyncio.current_task()
+        if self._task is task:
+            self._depth += 1
+            return True
+        await self._inner.acquire()
+        self._task = task
+        self._depth = 1
+        return True
+
+    def release(self) -> None:
+        if self._task is not asyncio.current_task():
+            raise RuntimeError("lock released by a non-owner")
+        self._depth -= 1
+        if self._depth == 0:
+            self._task = None
+            self._inner.release()
+
+    async def __aenter__(self) -> _TaskReentrantLock:
+        await self.acquire()
+        return self
+
+    async def __aexit__(self, *exc: object) -> None:
+        self.release()
+
+
+def _reentrant_connection_lock(database) -> _TaskReentrantLock:
+    """Return the shared connection lock, made task-reentrant.
+
+    ``CompletionEventsDao.append`` also acquires this lock, so
+    check+append in :func:`complete_instance` must re-enter on the
+    same task.  The wrapper is stored in the shared dict so the DAO
+    sees the same object.
+    """
+    loop = asyncio.get_running_loop()
+    key = (id(database), id(loop))
+    lock = _CONNECTION_LOCKS.get(key)
+    if isinstance(lock, _TaskReentrantLock):
+        return lock
+    wrapped = _TaskReentrantLock(
+        lock if lock is not None else asyncio.Lock()
+    )
+    _CONNECTION_LOCKS[key] = wrapped
+    return wrapped
+
+
 def _validate_int_id(value: object, field: str) -> int:
     """Reject non-int ids (bools included) before any lookup."""
     if isinstance(value, bool) or not isinstance(value, int):
@@ -214,7 +272,9 @@ async def complete_instance(
 
     Returns the derived state ``done``.  An already-done instance is a
     no-op: nothing is appended.  Unknown ``instance_id`` raises
-    ValueError naming the field.
+    ValueError naming the field.  Existence, latest-event, and the
+    append run under one connection lock so a concurrent complete
+    cannot double-append.
 
     ``was_on_time`` is recorded as True — a placeholder until the
     on-time task computes it from ``now`` / ``today`` against the
@@ -226,22 +286,31 @@ async def complete_instance(
     actor_source, actor_user_id, actor_child_id = _validate_actor(
         actor_source, actor_user_id, actor_child_id
     )
-    instance = await QuestInstancesDao(database).get_by_id(instance_id)
-    if instance is None:
-        raise ValueError(
-            f"instance_id: quest instance {instance_id} does not exist"
+    async with _reentrant_connection_lock(database):
+        instance = await QuestInstancesDao(database).get_by_id(instance_id)
+        if instance is None:
+            raise ValueError(
+                f"instance_id: quest instance {instance_id} does not exist"
+            )
+        latest = await CompletionEventsDao(database).get_latest_for_instance(
+            instance_id
+        )
+        if latest is not None and latest.event_type == EVENT_COMPLETED:
+            return "done"
+        if latest is not None and latest.event_type != EVENT_UNCOMPLETED:
+            raise ValueError(
+                f"event_type must be '{EVENT_COMPLETED}' or "
+                f"'{EVENT_UNCOMPLETED}', got {latest.event_type!r}"
+            )
+        await append_event(
+            database,
+            instance.id,
+            instance.child_id,
+            EVENT_COMPLETED,
+            actor_source=actor_source,
+            actor_user_id=actor_user_id,
+            actor_child_id=actor_child_id,
+            was_on_time=True,
+            now=now,
         )
-    if await instance_state(database, instance_id) == "done":
-        return "done"
-    await append_event(
-        database,
-        instance.id,
-        instance.child_id,
-        EVENT_COMPLETED,
-        actor_source=actor_source,
-        actor_user_id=actor_user_id,
-        actor_child_id=actor_child_id,
-        was_on_time=True,
-        now=now,
-    )
     return "done"