Run the missed sweep in the API and publish quest-missed on SSE (task 058c7b69) (#164)
This commit was merged in pull request #164.
This commit is contained in:
+12
-1
@@ -48,6 +48,7 @@ from api.database import DatabaseState, make_database
|
||||
from api.events import TransitionPublisher
|
||||
from api.routes_admin import router as admin_router
|
||||
from api.routes_panel import router as panel_router
|
||||
from api.scheduler import MissedSweepScheduler
|
||||
|
||||
|
||||
def _build_lifespan(db_path: str):
|
||||
@@ -57,7 +58,11 @@ def _build_lifespan(db_path: str):
|
||||
path in without threading it through FastAPI's lifespan signature.
|
||||
The lifespan also installs the app's ONE in-process transition
|
||||
publisher on ``app.state.publisher`` (see :mod:`api.events`) so
|
||||
every route and SSE subscriber shares that single instance.
|
||||
every route and SSE subscriber shares that single instance, and
|
||||
starts the daily missed-sweep scheduler
|
||||
(:class:`api.scheduler.MissedSweepScheduler`) against the same
|
||||
database and publisher — cancelled before the connection closes on
|
||||
shutdown.
|
||||
"""
|
||||
|
||||
@asynccontextmanager
|
||||
@@ -65,10 +70,16 @@ def _build_lifespan(db_path: str):
|
||||
state = make_database(db_path)
|
||||
app.state.db = state
|
||||
app.state.publisher = TransitionPublisher()
|
||||
scheduler = MissedSweepScheduler(
|
||||
state.database, publisher=app.state.publisher
|
||||
)
|
||||
app.state.sweep_scheduler = scheduler
|
||||
try:
|
||||
await state.open_and_migrate()
|
||||
scheduler.start()
|
||||
yield
|
||||
finally:
|
||||
await scheduler.stop()
|
||||
await state.close()
|
||||
|
||||
return lifespan
|
||||
|
||||
@@ -198,6 +198,12 @@ core_dao_children = importlib.import_module(
|
||||
#: delegating.
|
||||
core_dao_rules = importlib.import_module(f"{_CORE_MODULE_NAME}.dao_rules")
|
||||
|
||||
#: The HA-free missed sweep (``core.sweep.run_missed_sweep`` +
|
||||
#: ``SWEEP_WATERMARK_KEY``) the admin plane's trigger route calls and
|
||||
#: the daily scheduler runs — the ONE sweep policy the integration's
|
||||
#: bus shim shares.
|
||||
core_sweep = importlib.import_module(f"{_CORE_MODULE_NAME}.sweep")
|
||||
|
||||
__all__ = [
|
||||
"core",
|
||||
"core_children",
|
||||
@@ -218,4 +224,5 @@ __all__ = [
|
||||
"core_snapshot",
|
||||
"core_settings",
|
||||
"core_settings_store",
|
||||
"core_sweep",
|
||||
]
|
||||
|
||||
@@ -139,6 +139,23 @@ are deliberately not unified here:
|
||||
module's uniform None-as-absent rule). The response is the updated
|
||||
effective settings, the same shape ``GET`` returns.
|
||||
|
||||
The missed-sweep trigger (task 058c7b69) follows the same pattern over
|
||||
:mod:`nestquest_core.sweep` — the HA-free sweep policy the
|
||||
integration's rollover listener shares:
|
||||
|
||||
- ``POST /missed-sweep`` runs the sweep for the API host's local
|
||||
``today`` (ONE :func:`_local_now` clock read threaded into the core
|
||||
call) and publishes each returned ``(event_type, payload)`` pair
|
||||
DIRECTLY on the app's ONE transition publisher — the sweep-built
|
||||
payload, never re-fetched (the core commits its watermark before
|
||||
returning, so a failed re-fetch after it would lose the missed event
|
||||
forever), the same publish-what-the-core-built shape
|
||||
api/scheduler.py uses. The watermark rule
|
||||
(a same-day rerun announces nothing, a post-downtime run sweeps the
|
||||
accumulated window once), the no-completion-event rule and the
|
||||
read-only-over-the-domain-tables guarantee all live in the core; the
|
||||
response reports how many transitions were published.
|
||||
|
||||
Error mapping (every children, quest-definition and presence route,
|
||||
deliberately narrow):
|
||||
|
||||
@@ -237,6 +254,7 @@ from api.nestquest_core import (
|
||||
core_recurrence,
|
||||
core_settings,
|
||||
core_settings_store,
|
||||
core_sweep,
|
||||
)
|
||||
|
||||
#: All admin-plane routes share this router; the ONE admin dependency
|
||||
@@ -654,6 +672,17 @@ class AdminSettingsResponse(BaseModel):
|
||||
celebration_enabled: bool
|
||||
|
||||
|
||||
class AdminMissedSweepResponse(BaseModel):
|
||||
"""The result of ``POST /api/v1/admin/missed-sweep``.
|
||||
|
||||
``fired`` is the number of ``nestquest_quest_missed`` transitions
|
||||
the run built AND this route published on the SSE stream — zero
|
||||
when the sweep's watermark already covers today (a same-day rerun).
|
||||
"""
|
||||
|
||||
fired: int
|
||||
|
||||
|
||||
#: 404 detail for a child id the core layer reports as non-existent.
|
||||
_CHILD_NOT_FOUND_DETAIL = "Child not found"
|
||||
|
||||
@@ -803,6 +832,19 @@ def _raise_settings_error(error: ValueError) -> NoReturn:
|
||||
raise HTTPException(status_code=422, detail=str(error)) from error
|
||||
|
||||
|
||||
def _raise_sweep_error(error: ValueError) -> NoReturn:
|
||||
"""Map a core missed-sweep ``ValueError`` onto 422 and raise it.
|
||||
|
||||
The sweep route has no path ids, so it has no 404 case: a core
|
||||
``ValueError`` — a rejected date the sweep's query refuses before
|
||||
any read, unreachable from this handler because ``today`` comes
|
||||
from the route's own clock read — is 422 with the core's message
|
||||
as the detail. Only ``ValueError`` is caught, so an unexpected
|
||||
failure re-raises rather than becoming a 4xx.
|
||||
"""
|
||||
raise HTTPException(status_code=422, detail=str(error)) from error
|
||||
|
||||
|
||||
def _local_now() -> datetime.datetime:
|
||||
"""Return the API host's local time as ONE timezone-aware clock read.
|
||||
|
||||
@@ -1635,4 +1677,56 @@ async def admin_update_settings(
|
||||
return _settings_response(settings)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/missed-sweep",
|
||||
summary="Run the nightly missed-quest sweep now",
|
||||
response_model=AdminMissedSweepResponse,
|
||||
)
|
||||
async def admin_run_missed_sweep(
|
||||
request: Request,
|
||||
) -> AdminMissedSweepResponse:
|
||||
"""Run the missed-quest sweep and publish what it built.
|
||||
|
||||
Requires a valid admin JWT (the router's shared
|
||||
:func:`~api.auth.require_admin` dependency — the ONE check; this
|
||||
handler performs NO auth of its own). Thin adapter over the SAME
|
||||
HA-free sweep policy the integration's rollover listener runs
|
||||
(:mod:`nestquest_core.sweep`): ONE timezone-aware clock read
|
||||
(:func:`_local_now`) pins the API host's local ``today``, and ONE
|
||||
:func:`nestquest_core.sweep.run_missed_sweep` call does everything
|
||||
else — the past-due open-instance query, the documented payload
|
||||
build and the watermark update (a same-day rerun is an empty
|
||||
no-op, and a run after downtime sweeps the accumulated window
|
||||
once). The route is read-only over the domain tables: it never
|
||||
mutates an instance and never writes a completion event.
|
||||
|
||||
Each returned ``(event_type, payload)`` pair is published DIRECTLY
|
||||
on the app's ONE transition publisher (``request.app.state.publisher``,
|
||||
fan-out fire-and-forget, see api/events.py) — the exact payload the
|
||||
sweep built, never re-fetched and never hand-built here, the same
|
||||
publish-what-the-core-built shape api/scheduler.py uses. This is a
|
||||
correctness rule, not a style preference: the core commits the
|
||||
idempotency watermark BEFORE returning, so a re-fetch-based publish
|
||||
loop opens a loss window — a concurrent regenerate deleting the
|
||||
instance, or a transient database error, would raise uncaught (a
|
||||
bare 500) AFTER the watermark advanced, and no future sweep would
|
||||
ever re-examine that instance. Publishing the built payload leaves
|
||||
no such window, and every event of one run keeps the ONE shared
|
||||
``occurred_at`` stamp the sweep computed for it. The response
|
||||
reports ``fired`` — the number of transitions published. Error
|
||||
mapping (:func:`_raise_sweep_error`): every core ``ValueError`` is
|
||||
422; anything else re-raises.
|
||||
"""
|
||||
state: DatabaseState = request.app.state.db
|
||||
database = state.database
|
||||
today = _local_now().date()
|
||||
try:
|
||||
events = await core_sweep.run_missed_sweep(database, today=today)
|
||||
except ValueError as error:
|
||||
_raise_sweep_error(error)
|
||||
for event_type, payload in events:
|
||||
request.app.state.publisher.publish(event_type, payload)
|
||||
return AdminMissedSweepResponse(fired=len(events))
|
||||
|
||||
|
||||
__all__ = ["router"]
|
||||
|
||||
@@ -0,0 +1,173 @@
|
||||
"""Daily missed-sweep background scheduler for the API service.
|
||||
|
||||
The API service has no Home Assistant, so nobody re-fires the
|
||||
integration's day-rollover listener for it: this module owns the API
|
||||
plane's own background job. One :class:`MissedSweepScheduler` per app
|
||||
(installed on ``app.state`` by the lifespan) sleeps until the
|
||||
configured ``day_rollover_time`` — read fresh from the settings store
|
||||
at the top of EVERY cycle, so an admin's settings change applies on
|
||||
the next scheduling decision without a restart — then runs the
|
||||
HA-free core's :func:`nestquest_core.sweep.run_missed_sweep` for the
|
||||
API host's local date and publishes each returned
|
||||
``nestquest_quest_missed`` event on the app's ONE transition publisher
|
||||
(the same SSE stream every route publishes through).
|
||||
|
||||
All sweep policy lives in the core (the watermark, the
|
||||
``[watermark, today)`` window, the no-completion-event rule): this
|
||||
scheduler only decides WHEN. A same-night rerun — e.g. two workers,
|
||||
or a settings change moving the rollover past itself — fires nothing:
|
||||
the core's watermark makes the run an empty no-op.
|
||||
|
||||
Testability: ``clock`` (an aware-now callable) and ``sleep`` (an
|
||||
awaitable delay) are injectable, so tests pin the time, fast-forward
|
||||
through the wait, and observe the computed delays without waiting a
|
||||
real day. Cancellation: :meth:`stop` cancels the loop task and waits
|
||||
for it, so the lifespan's shutdown is clean and a mid-run sweep is
|
||||
interrupted before the connection closes. A failed iteration (a
|
||||
database hiccup, a rejected setting) is logged and retried after a
|
||||
fixed backoff rather than silently killing the job.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import datetime
|
||||
import logging
|
||||
|
||||
from api.nestquest_core import core_settings_store, core_sweep
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
#: Backoff after a failed scheduling cycle (settings read or sweep
|
||||
#: error), so a persistent failure retries every minute instead of
|
||||
#: spinning or dying.
|
||||
_RETRY_BACKOFF_SECONDS = 60.0
|
||||
|
||||
|
||||
def _delay_seconds(target: datetime.time, now: datetime.datetime) -> float:
|
||||
"""Seconds from ``now`` until the next occurrence of ``target``.
|
||||
|
||||
``target`` is a plain ``HH:MM`` time of day: the next wall-clock
|
||||
occurrence strictly AFTER ``now`` (an equal time schedules for
|
||||
tomorrow, so the just-run sweep is never instantly re-run).
|
||||
"""
|
||||
scheduled = now.replace(
|
||||
hour=target.hour, minute=target.minute, second=0, microsecond=0
|
||||
)
|
||||
if scheduled <= now:
|
||||
scheduled += datetime.timedelta(days=1)
|
||||
return (scheduled - now).total_seconds()
|
||||
|
||||
|
||||
class MissedSweepScheduler:
|
||||
"""Run the missed sweep once per day at ``day_rollover_time``."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
database: object,
|
||||
*,
|
||||
publisher: object | None = None,
|
||||
clock: object | None = None,
|
||||
sleep: object | None = None,
|
||||
) -> None:
|
||||
"""Create the scheduler (not yet running).
|
||||
|
||||
``database`` is the app's live :class:`~api.database.NestQuestDatabase`.
|
||||
``publisher`` is the app's ONE
|
||||
:class:`~api.events.TransitionPublisher` (publishing is skipped
|
||||
when ``None``). ``clock`` defaults to the API host's local
|
||||
aware now (``datetime.datetime.now().astimezone()``) — the same
|
||||
single-clock discipline the routes' ``_local_now`` uses;
|
||||
``sleep`` defaults to :func:`asyncio.sleep`.
|
||||
"""
|
||||
self._database = database
|
||||
self._publisher = publisher
|
||||
self._clock = clock if clock is not None else (
|
||||
lambda: datetime.datetime.now().astimezone()
|
||||
)
|
||||
self._sleep = sleep if sleep is not None else asyncio.sleep
|
||||
self._task: asyncio.Task | None = None
|
||||
|
||||
def start(self) -> None:
|
||||
"""Start the background loop (idempotent: a running one stays)."""
|
||||
if self._task is None or self._task.done():
|
||||
self._task = asyncio.create_task(
|
||||
self._run_forever(), name="nestquest-missed-sweep"
|
||||
)
|
||||
|
||||
async def stop(self) -> None:
|
||||
"""Cancel the loop and wait for it to finish (safe if never started)."""
|
||||
if self._task is not None:
|
||||
self._task.cancel()
|
||||
try:
|
||||
await self._task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self._task = None
|
||||
|
||||
@property
|
||||
def running(self) -> bool:
|
||||
"""Whether the background loop task is alive."""
|
||||
return self._task is not None and not self._task.done()
|
||||
|
||||
async def _run_forever(self) -> None:
|
||||
"""Schedule → sweep → repeat, forever (until cancelled)."""
|
||||
while True:
|
||||
try:
|
||||
settings = await core_settings_store.load_settings(
|
||||
self._database
|
||||
)
|
||||
rollover = datetime.time.fromisoformat(
|
||||
settings.day_rollover_time
|
||||
)
|
||||
except Exception:
|
||||
# Not silent: the failure is logged and retried after
|
||||
# the backoff, so one bad settings document or a
|
||||
# transient database error never kills the scheduler.
|
||||
LOGGER.exception(
|
||||
"NestQuest missed-sweep scheduler failed to resolve "
|
||||
"the day rollover time; retrying in %gs",
|
||||
_RETRY_BACKOFF_SECONDS,
|
||||
)
|
||||
await self._sleep(_RETRY_BACKOFF_SECONDS)
|
||||
continue
|
||||
delay = _delay_seconds(rollover, self._clock())
|
||||
LOGGER.info(
|
||||
"NestQuest missed sweep scheduled for %s (%.0fs from now)",
|
||||
settings.day_rollover_time,
|
||||
delay,
|
||||
)
|
||||
await self._sleep(delay)
|
||||
try:
|
||||
await self._run_once(self._clock().date())
|
||||
except Exception:
|
||||
# Not silent either: the sweep error is logged and the
|
||||
# loop backs off before the next cycle, whose watermark
|
||||
# re-read makes an already-swept day a no-op.
|
||||
LOGGER.exception(
|
||||
"NestQuest scheduled missed sweep failed; retrying in "
|
||||
"%gs",
|
||||
_RETRY_BACKOFF_SECONDS,
|
||||
)
|
||||
await self._sleep(_RETRY_BACKOFF_SECONDS)
|
||||
|
||||
async def _run_once(self, today: datetime.date) -> None:
|
||||
"""Run ONE sweep for ``today`` and publish what it built.
|
||||
|
||||
A failure propagates to :meth:`_run_forever`, which logs it and
|
||||
backs off before the next cycle; the watermark makes a re-run
|
||||
of an already-swept day announce nothing twice.
|
||||
"""
|
||||
events = await core_sweep.run_missed_sweep(
|
||||
self._database, today=today
|
||||
)
|
||||
for event_type, payload in events:
|
||||
if self._publisher is not None:
|
||||
self._publisher.publish(event_type, payload)
|
||||
LOGGER.info(
|
||||
"NestQuest scheduled missed sweep for %s published %d event(s)",
|
||||
today.isoformat(),
|
||||
len(events),
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["MissedSweepScheduler"]
|
||||
@@ -0,0 +1,143 @@
|
||||
"""The nightly missed-quest sweep, Home-Assistant-free (Feature 11).
|
||||
|
||||
This is the HA-free heart of the missed sweep, shared by the two
|
||||
planes: the Home Assistant integration's :mod:`~..sweep` shim (which
|
||||
fires the returned events on ``hass.bus``) and the API service (which
|
||||
publishes them on the SSE stream and can also trigger the run on
|
||||
demand). It owns the ONE sweep policy:
|
||||
|
||||
At the configured day-rollover time (or on an explicit trigger) this
|
||||
job finds every instance whose ``due_date`` is before the
|
||||
household-local date, that has NO completion event, and that falls
|
||||
after the persisted sweep watermark — and RETURNS one
|
||||
``nestquest_quest_missed`` event per instance with the documented
|
||||
payload. Like every core module, it never touches the Home Assistant
|
||||
event bus and never mutates an instance: ``missed`` is derived state,
|
||||
never an event_type (Feature 08's contract), so the sweep is read-only
|
||||
over the domain tables.
|
||||
|
||||
Idempotency is the watermark: after a run, ``nestquest_meta_state``
|
||||
records the household-local date the sweep ran, and the next run only
|
||||
sweeps instances due in ``[watermark, today)``. A second run the same
|
||||
night therefore fires nothing new, and a run after days of downtime
|
||||
sweeps exactly the accumulated window once — no per-instance journal,
|
||||
no re-announcing old misses. An instance swept as missed that is
|
||||
later completed simply gains its completion event; history keeps both
|
||||
rows (append-only), and the sweep's later runs skip it via the
|
||||
no-completion-event rule.
|
||||
|
||||
The whole watermark read → query → build → write span runs inside the
|
||||
per-database sweep lock (one lock per connection wrapper and running
|
||||
loop, the settings store's pattern): an overlapping startup sweep and
|
||||
a rollover sweep queue instead of interleaving, and the loser re-reads
|
||||
the watermark INSIDE the lock — sees the winner's date — and announces
|
||||
nothing twice.
|
||||
|
||||
The caller supplies ``today`` as a plain :class:`datetime.date`: the
|
||||
household-local calendar date, read from ``hass.config.time_zone`` by
|
||||
the integration or from the API host's local clock by the API. This
|
||||
module never reads a timezone database or config of its own.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import datetime
|
||||
import logging
|
||||
|
||||
from .const import EVENT_QUEST_MISSED
|
||||
from .dao_meta import MetaStateDao
|
||||
from .dao_instances import QuestInstancesDao
|
||||
from .db import NestQuestDatabase
|
||||
# Intentional intra-package reuse of the shared payload builder: the
|
||||
# missed event's payload is the documented §3 instance payload, and
|
||||
# building it through the ONE builder keeps this sweep's payloads from
|
||||
# drifting from the completion/uncompleted builders' shape.
|
||||
from .events import _instance_payload
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
#: meta_state key holding the household-local date the sweep last ran.
|
||||
SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
|
||||
|
||||
#: One asyncio.Lock per (connection wrapper, running loop), mirroring
|
||||
#: the settings store's pattern: the whole watermark read → query →
|
||||
#: build → write span is one critical section, so an overlapping
|
||||
#: startup sweep and rollover sweep queue instead of double-announcing.
|
||||
#: The loser re-reads the watermark INSIDE the lock, sees the winner's
|
||||
#: date, and fires nothing.
|
||||
_SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
|
||||
|
||||
|
||||
def _sweep_lock(database: NestQuestDatabase) -> asyncio.Lock:
|
||||
"""Return the sweep lock bound to ``database``'s running loop."""
|
||||
loop = asyncio.get_running_loop()
|
||||
key = (id(database), id(loop))
|
||||
lock = _SWEEP_LOCKS.get(key)
|
||||
if lock is None:
|
||||
lock = asyncio.Lock()
|
||||
_SWEEP_LOCKS[key] = lock
|
||||
return lock
|
||||
|
||||
|
||||
#: Strict UTC ISO-8601 timestamp (the DAO layers' policy): one shape,
|
||||
#: explicit offset, second precision.
|
||||
_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
|
||||
|
||||
|
||||
async def run_missed_sweep(
|
||||
database: NestQuestDatabase,
|
||||
*,
|
||||
today: datetime.date,
|
||||
) -> list[tuple[str, dict]]:
|
||||
"""Build one missed event per unswept past-due open instance.
|
||||
|
||||
``today`` is the household-local calendar date the sweep runs for
|
||||
(the caller owns the clock: the integration reads it from
|
||||
``hass.config.time_zone``, the API from its host's local time).
|
||||
Returns the ``(event_type, payload)`` tuples to announce, EMPTY
|
||||
when the watermark already covers ``today`` (a same-night rerun) —
|
||||
the caller decides how to publish them and announces nothing on
|
||||
its own.
|
||||
|
||||
The whole watermark read → query → build → write span runs inside
|
||||
the per-database sweep lock: an overlapping startup sweep and a
|
||||
rollover sweep queue instead of interleaving, and the loser
|
||||
re-reads the watermark inside the lock — sees the winner's date —
|
||||
and announces nothing twice. Every event in one run shares one
|
||||
``occurred_at`` stamp (one run, one announcement moment), and the
|
||||
payload comes from the shared
|
||||
:func:`~.events._instance_payload` builder so the shape can never
|
||||
drift from the documented §3 contract.
|
||||
"""
|
||||
today_iso = today.isoformat()
|
||||
dao = MetaStateDao(database)
|
||||
async with _sweep_lock(database):
|
||||
watermark = await dao.get(SWEEP_WATERMARK_KEY)
|
||||
if watermark == today_iso:
|
||||
LOGGER.debug(
|
||||
"NestQuest missed sweep already ran for %s", today_iso
|
||||
)
|
||||
return []
|
||||
|
||||
instances_dao = QuestInstancesDao(database)
|
||||
rows = await instances_dao.list_missed_unswept(today_iso, watermark)
|
||||
now_stamp = datetime.datetime.now(datetime.timezone.utc).strftime(
|
||||
_UTC_TIMESTAMP_FORMAT
|
||||
)
|
||||
events: list[tuple[str, dict]] = []
|
||||
for instance in rows:
|
||||
events.append(
|
||||
(
|
||||
EVENT_QUEST_MISSED,
|
||||
await _instance_payload(
|
||||
database, instance.id, now_stamp
|
||||
),
|
||||
)
|
||||
)
|
||||
await dao.set(SWEEP_WATERMARK_KEY, today_iso)
|
||||
LOGGER.info(
|
||||
"NestQuest missed sweep for %s built %d event(s)",
|
||||
today_iso,
|
||||
len(events),
|
||||
)
|
||||
return events
|
||||
@@ -1,70 +1,40 @@
|
||||
"""The nightly missed-quest sweep (Feature 11).
|
||||
"""The nightly missed-quest sweep (Feature 11) — the Home Assistant shim.
|
||||
|
||||
At the configured day-rollover time (the same HA-local listener that
|
||||
re-materializes the rolling horizon) this job finds every instance
|
||||
whose ``due_date`` is before the HA-local date, that has NO completion
|
||||
event, and that falls after the persisted sweep watermark — and fires
|
||||
ONE ``nestquest_quest_missed`` event per instance with the documented
|
||||
payload. The sweep is read-only over the domain tables: it never
|
||||
mutates an instance and never writes a completion event (``missed``
|
||||
is derived state, never an event_type — Feature 08's contract).
|
||||
The sweep's policy is the HA-free core's :mod:`~.core.sweep`: the
|
||||
watermark read → ``list_missed_unswept`` query → payload build →
|
||||
watermark span, all inside the per-database sweep lock, shared with
|
||||
the API plane. This module is the integration's thin adapter over it:
|
||||
at the configured day-rollover time (the same HA-local listener that
|
||||
re-materializes the rolling horizon) and once at setup it calls the
|
||||
core's :func:`~.core.sweep.run_missed_sweep` and fires each returned
|
||||
event on ``hass.bus`` — ONE ``nestquest_quest_missed`` per past-due
|
||||
open instance with the documented payload. The core never touches
|
||||
Home Assistant; the firing lives here.
|
||||
|
||||
Idempotency is the watermark: after a run, ``nestquest_meta_state``
|
||||
records the HA-local date the sweep ran, and the next run only
|
||||
sweeps instances due in ``[watermark, today)``. A second run the
|
||||
same night therefore fires nothing new, and a run after days of
|
||||
downtime sweeps exactly the accumulated window once — no
|
||||
per-instance journal, no re-announcing old misses. An instance swept
|
||||
as missed that is later completed simply gains its completion event;
|
||||
history keeps both rows (append-only), and the sweep's later runs
|
||||
skip it via the no-completion-event rule.
|
||||
Idempotency is the core's watermark: after a run, ``nestquest_meta_state``
|
||||
records the HA-local date the sweep ran, and the next run only sweeps
|
||||
instances due in ``[watermark, today)`` — a second run the same night
|
||||
fires nothing new, and a run after days of downtime sweeps exactly the
|
||||
accumulated window once. An instance swept as missed that is later
|
||||
completed simply gains its completion event; the sweep is read-only
|
||||
over the domain tables (``missed`` is derived state, never an
|
||||
event_type — Feature 08's contract).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import datetime
|
||||
import logging
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
from .const import EVENT_QUEST_MISSED
|
||||
from .dao_children import ChildrenDao
|
||||
from .dao_instances import QuestInstancesDao
|
||||
from .dao_meta import MetaStateDao
|
||||
from .dao_rules import QuestDefinitionsDao
|
||||
from .core.sweep import (
|
||||
SWEEP_WATERMARK_KEY,
|
||||
run_missed_sweep as _run_core_missed_sweep,
|
||||
)
|
||||
from .db import NestQuestDatabase
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
#: meta_state key holding the HA-local date the sweep last ran.
|
||||
SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
|
||||
|
||||
#: One asyncio.Lock per (connection wrapper, running loop), mirroring
|
||||
#: the migration runner's pattern: the whole watermark read → query →
|
||||
#: fire → write span is one critical section, so an overlapping startup
|
||||
#: sweep and rollover sweep queue instead of double-announcing. The
|
||||
#: loser re-reads the watermark INSIDE the lock, sees the winner's
|
||||
#: date, and fires nothing.
|
||||
_SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
|
||||
|
||||
|
||||
def _sweep_lock(database: NestQuestDatabase) -> asyncio.Lock:
|
||||
"""Return the sweep lock bound to ``database``'s running loop."""
|
||||
loop = asyncio.get_running_loop()
|
||||
key = (id(database), id(loop))
|
||||
lock = _SWEEP_LOCKS.get(key)
|
||||
if lock is None:
|
||||
lock = asyncio.Lock()
|
||||
_SWEEP_LOCKS[key] = lock
|
||||
return lock
|
||||
|
||||
#: UTC ISO-8601 timestamp policy shared with the DAO layers.
|
||||
_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
|
||||
|
||||
|
||||
def _now_stamp() -> str:
|
||||
return datetime.datetime.now(datetime.timezone.utc).strftime(
|
||||
_UTC_TIMESTAMP_FORMAT
|
||||
)
|
||||
__all__ = ["SWEEP_WATERMARK_KEY", "run_missed_sweep"]
|
||||
|
||||
|
||||
async def run_missed_sweep(
|
||||
@@ -79,54 +49,24 @@ async def run_missed_sweep(
|
||||
materialization and completion layers); when omitted it is read
|
||||
from ``hass.config.time_zone``. Returns the number of events
|
||||
fired. Fires nothing and returns 0 when the watermark already
|
||||
covers ``today`` (a same-night rerun).
|
||||
|
||||
The whole watermark read → query → fire → write span runs inside
|
||||
the per-database sweep lock: an overlapping startup sweep and a
|
||||
rollover sweep queue instead of interleaving, and the loser
|
||||
re-reads the watermark inside the lock — sees the winner's date —
|
||||
and announces nothing twice.
|
||||
covers ``today`` (a same-night rerun) — the watermark read, the
|
||||
query, the payload build and the watermark write all live in the
|
||||
core's locked
|
||||
:func:`~.core.sweep.run_missed_sweep`; this shim only fires each
|
||||
returned event on the bus and counts them.
|
||||
"""
|
||||
today_date = (
|
||||
today
|
||||
if today is not None
|
||||
else datetime.datetime.now(ZoneInfo(hass.config.time_zone)).date()
|
||||
)
|
||||
today_iso = today_date.isoformat()
|
||||
dao = MetaStateDao(database)
|
||||
async with _sweep_lock(database):
|
||||
watermark = await dao.get(SWEEP_WATERMARK_KEY)
|
||||
if watermark == today_iso:
|
||||
LOGGER.debug(
|
||||
"NestQuest missed sweep already ran for %s", today_iso
|
||||
)
|
||||
return 0
|
||||
|
||||
instances_dao = QuestInstancesDao(database)
|
||||
rows = await instances_dao.list_missed_unswept(today_iso, watermark)
|
||||
children_dao = ChildrenDao(database)
|
||||
definitions_dao = QuestDefinitionsDao(database)
|
||||
now_stamp = _now_stamp()
|
||||
fired = 0
|
||||
for instance in rows:
|
||||
child = await children_dao.get(instance.child_id)
|
||||
definition = await definitions_dao.get(instance.definition_id)
|
||||
hass.bus.async_fire(
|
||||
EVENT_QUEST_MISSED,
|
||||
{
|
||||
"child_id": instance.child_id,
|
||||
"child_name": child.display_name if child else None,
|
||||
"instance_id": instance.id,
|
||||
"quest_title": definition.title if definition else None,
|
||||
"window": instance.window,
|
||||
"due_date": instance.due_date,
|
||||
"due_time": instance.due_time,
|
||||
"occurred_at": now_stamp,
|
||||
},
|
||||
)
|
||||
fired += 1
|
||||
await dao.set(SWEEP_WATERMARK_KEY, today_iso)
|
||||
events = await _run_core_missed_sweep(database, today=today_date)
|
||||
for event_type, payload in events:
|
||||
hass.bus.async_fire(event_type, payload)
|
||||
fired = len(events)
|
||||
LOGGER.info(
|
||||
"NestQuest missed sweep for %s fired %d event(s)", today_iso, fired
|
||||
"NestQuest missed sweep for %s fired %d event(s)",
|
||||
today_date.isoformat(),
|
||||
fired,
|
||||
)
|
||||
return fired
|
||||
@@ -0,0 +1,671 @@
|
||||
"""Admin-plane missed-sweep trigger tests (task 058c7b69).
|
||||
|
||||
These tests exercise ``POST /api/v1/admin/missed-sweep`` — the admin
|
||||
plane's trigger for the nightly missed-quest sweep — against a REAL
|
||||
server (uvicorn on an ephemeral port, the SSE tests' documented
|
||||
approach: httpx's ASGI transport buffers whole bodies, so a live
|
||||
stream needs a live server), plus the API service's daily scheduler:
|
||||
|
||||
- The trigger requires the router's ONE ``require_admin`` check: a
|
||||
non-admin JWT is 403 and an absent credential is 401, and neither
|
||||
refusal runs the sweep.
|
||||
- Seeded past-due open instances: ONE ``nestquest_quest_missed``
|
||||
transition per instance arrives on the LIVE SSE stream with the
|
||||
documented payload fields (child_id, child_name, instance_id,
|
||||
quest_title, window, due_date, due_time, occurred_at — §3), while
|
||||
the seeded TODAY-due instance publishes nothing (the sweep's rule
|
||||
is ``due_date`` strictly before ``today``).
|
||||
- A same-day rerun publishes nothing (the core watermark).
|
||||
- A run after downtime (``today`` advanced two days) publishes the
|
||||
accumulated window once — the instance due on day one fires on the
|
||||
catch-up run — and a further rerun is silent again.
|
||||
- Publishing is the api/scheduler.py shape — the sweep-BUILT
|
||||
``(event_type, payload)`` pairs go straight to the publisher, no
|
||||
per-event re-fetch: all events of one run share the run's ONE
|
||||
``occurred_at`` stamp, and a deterministic concurrent-deletion case
|
||||
(the sweep is wrapped to delete one swept instance after it built
|
||||
its events and committed its watermark) still answers 200 and
|
||||
streams every built event — a re-fetch there would 500 and lose the
|
||||
missed event forever.
|
||||
- The daily scheduler (:class:`api.scheduler.MissedSweepScheduler`)
|
||||
computes the delay to the configured ``day_rollover_time``, runs the
|
||||
sweep when the (injectable) clock reaches it, publishes the built
|
||||
events on its publisher, advances to the NEXT day's rollover, and
|
||||
cancels cleanly.
|
||||
|
||||
The household is seeded through the SAME core copy the app uses (the
|
||||
``nestquest_core.*`` package registered by ``api/nestquest_core.py``;
|
||||
see the two-copy caveat there) and the sweep's watermark is read back
|
||||
through the same copy's :class:`~nestquest_core.dao_meta.MetaStateDao`
|
||||
— on the SERVER's event loop for the uvicorn-hosted app (the database
|
||||
belongs to that thread's loop, so assertions schedule there), on the
|
||||
test's own loop for the scheduler test's standalone database.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import datetime
|
||||
import importlib
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import uvicorn
|
||||
|
||||
from api import routes_admin
|
||||
from api.app import create_app
|
||||
from api.auth import JwksCache
|
||||
from api.config import ApiConfig
|
||||
from api.database import _executor
|
||||
from api.nestquest_core import core_db as _core_db
|
||||
from api.nestquest_core import core_migrations as _core_migrations
|
||||
from api.scheduler import MissedSweepScheduler
|
||||
from tests.admin_jwt_harness import (
|
||||
ADMIN_KEY,
|
||||
AUDIENCE,
|
||||
ISSUER,
|
||||
JWKS_URL,
|
||||
KID,
|
||||
StubbedFetch,
|
||||
jwks_for,
|
||||
make_token,
|
||||
)
|
||||
from tests.test_api_panel import PANEL_TOKEN
|
||||
from tests.test_api_sse import EVENT_QUEST_MISSED, READ_TIMEOUT, _free_port, _stream
|
||||
|
||||
# The bundled core submodules, through the SAME copy the app uses
|
||||
# (nestquest_core.*, NOT the conftest-stubbed custom_components copy).
|
||||
_children = importlib.import_module("nestquest_core.children")
|
||||
_quest_definitions = importlib.import_module(
|
||||
"nestquest_core.quest_definitions"
|
||||
)
|
||||
_materialize = importlib.import_module("nestquest_core.materialize")
|
||||
_recurrence = importlib.import_module("nestquest_core.recurrence")
|
||||
_dao_instances = importlib.import_module("nestquest_core.dao_instances")
|
||||
_dao_meta = importlib.import_module("nestquest_core.dao_meta")
|
||||
_settings_store = importlib.import_module("nestquest_core.settings_store")
|
||||
_core_sweep = importlib.import_module("nestquest_core.sweep")
|
||||
|
||||
#: The well-known ``nestquest_meta_state`` key holding the sweep
|
||||
#: watermark — asserted literally so the contract is pinned here.
|
||||
SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
|
||||
|
||||
|
||||
def _admin_headers() -> dict[str, str]:
|
||||
"""The Authorization header of a valid nestquest-admins JWT."""
|
||||
return {"Authorization": f"Bearer {make_token()}"}
|
||||
|
||||
|
||||
async def _seed_missed_household(
|
||||
db_path: str, today: datetime.date
|
||||
) -> SimpleNamespace:
|
||||
"""Seed the missed-sweep household on ``db_path`` and return records.
|
||||
|
||||
Two children; two quests whose ONLY instances are due YESTERDAY
|
||||
(created via ``materialize`` with ``today`` pinned to yesterday —
|
||||
the no-past guard accepts the rows, and they land past-due with no
|
||||
completion events: the sweep's targets), and one quest due TODAY
|
||||
(not past-due: the sweep must NOT announce it). The yesterday
|
||||
rules carry ``end_date`` so the later today-walk cannot give them
|
||||
a second instance. Opens, migrates, seeds and CLOSES its own
|
||||
connection so the app's lifespan connection never overlaps it.
|
||||
"""
|
||||
database = _core_db.NestQuestDatabase(_executor)
|
||||
await database.open(db_path)
|
||||
try:
|
||||
await _core_migrations.apply_migrations(database)
|
||||
today_iso = today.isoformat()
|
||||
yesterday = today - datetime.timedelta(days=1)
|
||||
yesterday_iso = yesterday.isoformat()
|
||||
|
||||
ada = await _children.create_child(database, "Ada", sort_order=0)
|
||||
bo = await _children.create_child(database, "Bo", sort_order=1)
|
||||
|
||||
yesterday_rule = _recurrence.ScheduleRule.from_dict(
|
||||
{
|
||||
"rule_type": "daily",
|
||||
"start_date": yesterday_iso,
|
||||
"end_date": yesterday_iso,
|
||||
}
|
||||
)
|
||||
await _quest_definitions.create_quest_definition(
|
||||
database,
|
||||
"Stale chore",
|
||||
yesterday_rule,
|
||||
[ada.id],
|
||||
[("morning", "09:00")],
|
||||
)
|
||||
await _quest_definitions.create_quest_definition(
|
||||
database,
|
||||
"Old chore",
|
||||
yesterday_rule,
|
||||
[bo.id],
|
||||
[("evening", "18:30")],
|
||||
)
|
||||
# Past-due open instances for the two yesterday quests (the
|
||||
# walk pins ``today`` to yesterday so the rows are accepted).
|
||||
await _materialize.materialize(
|
||||
database, yesterday_iso, yesterday_iso, today=yesterday
|
||||
)
|
||||
# Today's quest: created after the yesterday walk so it has no
|
||||
# past-due instance, then materialized for today only.
|
||||
await _quest_definitions.create_quest_definition(
|
||||
database,
|
||||
"Recoverable chore",
|
||||
_recurrence.ScheduleRule.from_dict(
|
||||
{"rule_type": "daily", "start_date": today_iso}
|
||||
),
|
||||
[ada.id],
|
||||
[("afternoon", "17:00")],
|
||||
)
|
||||
await _materialize.materialize(
|
||||
database, today_iso, today_iso, today=today
|
||||
)
|
||||
|
||||
instances = _dao_instances.QuestInstancesDao(database)
|
||||
ada_yesterday = await instances.list_by_child_and_date(
|
||||
ada.id, yesterday_iso
|
||||
)
|
||||
assert len(ada_yesterday) == 1
|
||||
stale = ada_yesterday[0]
|
||||
bo_yesterday = await instances.list_by_child_and_date(
|
||||
bo.id, yesterday_iso
|
||||
)
|
||||
assert len(bo_yesterday) == 1
|
||||
old = bo_yesterday[0]
|
||||
ada_today = await instances.list_by_child_and_date(ada.id, today_iso)
|
||||
assert len(ada_today) == 1
|
||||
recoverable = ada_today[0]
|
||||
return SimpleNamespace(
|
||||
ada=ada,
|
||||
bo=bo,
|
||||
stale_instance=stale,
|
||||
old_instance=old,
|
||||
recoverable_instance=recoverable,
|
||||
yesterday_iso=yesterday_iso,
|
||||
today_iso=today_iso,
|
||||
)
|
||||
finally:
|
||||
await database.close()
|
||||
|
||||
|
||||
class _AdminServer:
|
||||
"""Run the app with uvicorn on an ephemeral port, in a thread.
|
||||
|
||||
The SSE tests' documented approach (see tests/test_api_sse.py):
|
||||
uvicorn's factory mode runs ``create_app`` INSIDE the server
|
||||
thread, so the lifespan — database, publisher, the daily sweep
|
||||
scheduler — belongs to that thread's event loop. The app's JWKS
|
||||
cache is swapped for one built around the stubbed fetcher BEFORE
|
||||
the server starts, so admin JWTs genuinely verify with no network.
|
||||
"""
|
||||
|
||||
def __init__(self, db_path: str) -> None:
|
||||
self.port = _free_port()
|
||||
self._server_loop: asyncio.AbstractEventLoop | None = None
|
||||
self._server = uvicorn.Server(
|
||||
uvicorn.Config(
|
||||
self._make_app_factory(db_path),
|
||||
host="127.0.0.1",
|
||||
port=self.port,
|
||||
log_level="warning",
|
||||
)
|
||||
)
|
||||
self._thread = threading.Thread(target=self._server.run, daemon=True)
|
||||
|
||||
def _make_app_factory(self, db_path: str):
|
||||
"""The uvicorn app factory, closing over ``db_path``."""
|
||||
|
||||
def factory():
|
||||
app = create_app(
|
||||
ApiConfig(
|
||||
db_path=db_path,
|
||||
panel_token=PANEL_TOKEN,
|
||||
oidc_issuer=ISSUER,
|
||||
oidc_audience=AUDIENCE,
|
||||
oidc_jwks_url=JWKS_URL,
|
||||
)
|
||||
)
|
||||
app.state.jwks_cache = JwksCache(
|
||||
JWKS_URL, fetcher=StubbedFetch(jwks_for(ADMIN_KEY, KID))
|
||||
)
|
||||
self._server_loop = asyncio.get_running_loop()
|
||||
self.app = app
|
||||
return app
|
||||
|
||||
return factory
|
||||
|
||||
def __enter__(self) -> "_AdminServer":
|
||||
self._thread.start()
|
||||
deadline = time.monotonic() + READ_TIMEOUT
|
||||
while not self._server.started:
|
||||
if time.monotonic() > deadline or not self._thread.is_alive():
|
||||
raise RuntimeError("uvicorn did not start")
|
||||
time.sleep(0.01)
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc) -> None:
|
||||
self._server.should_exit = True
|
||||
self._thread.join(timeout=READ_TIMEOUT)
|
||||
|
||||
def url(self, path: str) -> str:
|
||||
"""The full URL for ``path`` on this server."""
|
||||
return f"http://127.0.0.1:{self.port}{path}"
|
||||
|
||||
def run_on_loop(self, coro) -> object:
|
||||
"""Run ``coro`` on the SERVER's event loop and wait for it.
|
||||
|
||||
The app (and its database) lives on the uvicorn thread's loop,
|
||||
so database assertions schedule there. Bounded by
|
||||
READ_TIMEOUT so a wedged schedule fails instead of hanging.
|
||||
"""
|
||||
return asyncio.run_coroutine_threadsafe(
|
||||
coro, self._server_loop
|
||||
).result(timeout=READ_TIMEOUT)
|
||||
|
||||
|
||||
def _watermark(server: _AdminServer) -> str | None:
|
||||
"""The stored sweep watermark, read on the server's own loop."""
|
||||
database = server.app.state.db.database
|
||||
return server.run_on_loop(
|
||||
_dao_meta.MetaStateDao(database).get(SWEEP_WATERMARK_KEY)
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db_path(tmp_path: Path) -> str:
|
||||
"""A fresh per-test SQLite path under pytest's tmp_path."""
|
||||
return str(tmp_path / "nestquest.db")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def missed_app(temp_db_path: str) -> SimpleNamespace:
|
||||
"""The seeded household served by a live uvicorn app, clock pinned.
|
||||
|
||||
``routes_admin._local_now`` is pinned to today 12:00 (the same
|
||||
pinning discipline the panel/SSE tests use) BEFORE the server
|
||||
starts, so the trigger route's ``today`` matches the seed; the
|
||||
downtime test moves the pin forward mid-test. Restored in
|
||||
``finally`` even on failure.
|
||||
"""
|
||||
now_local = datetime.datetime.now().astimezone()
|
||||
pinned_now = now_local.replace(hour=12, minute=0, second=0, microsecond=0)
|
||||
today = pinned_now.date()
|
||||
seed = await _seed_missed_household(temp_db_path, today)
|
||||
original = routes_admin._local_now
|
||||
routes_admin._local_now = lambda: pinned_now
|
||||
try:
|
||||
with _AdminServer(temp_db_path) as server:
|
||||
client = httpx.AsyncClient(base_url=server.url(""))
|
||||
async with client:
|
||||
yield SimpleNamespace(
|
||||
client=client,
|
||||
seed=seed,
|
||||
server=server,
|
||||
today=today,
|
||||
pinned_now=pinned_now,
|
||||
)
|
||||
finally:
|
||||
routes_admin._local_now = original
|
||||
|
||||
|
||||
# --- the trigger: one missed transition per past-due open instance -----
|
||||
|
||||
|
||||
async def test_trigger_publishes_one_missed_event_per_instance(
|
||||
missed_app,
|
||||
) -> None:
|
||||
"""The trigger streams one documented missed transition per instance.
|
||||
|
||||
The stream is subscribed FIRST, the sweep is triggered through the
|
||||
route as an admin, and each emitted frame's type and every
|
||||
documented payload field are asserted; the TODAY-due instance must
|
||||
NOT appear (the sweep's rule is strictly past-due).
|
||||
"""
|
||||
ns = missed_app
|
||||
seed = ns.seed
|
||||
stream = _stream(ns.server.url("/api/v1/panel/events"))
|
||||
async with stream:
|
||||
response = await ns.client.post(
|
||||
"/api/v1/admin/missed-sweep", headers=_admin_headers()
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"fired": 2}
|
||||
|
||||
event_type, payload = await stream.next_event()
|
||||
assert event_type == EVENT_QUEST_MISSED
|
||||
assert event_type == "nestquest_quest_missed"
|
||||
assert payload["child_id"] == seed.ada.id
|
||||
assert payload["child_name"] == "Ada"
|
||||
assert payload["instance_id"] == seed.stale_instance.id
|
||||
assert payload["quest_title"] == "Stale chore"
|
||||
assert payload["window"] == "morning"
|
||||
assert payload["due_date"] == seed.yesterday_iso
|
||||
assert payload["due_time"] == "09:00"
|
||||
assert payload["occurred_at"] # strict UTC ISO-8601 stamp
|
||||
run_stamp = payload["occurred_at"]
|
||||
|
||||
event_type, payload = await stream.next_event()
|
||||
assert event_type == EVENT_QUEST_MISSED
|
||||
assert payload["child_id"] == seed.bo.id
|
||||
assert payload["child_name"] == "Bo"
|
||||
assert payload["instance_id"] == seed.old_instance.id
|
||||
assert payload["quest_title"] == "Old chore"
|
||||
assert payload["window"] == "evening"
|
||||
assert payload["due_date"] == seed.yesterday_iso
|
||||
assert payload["due_time"] == "18:30"
|
||||
# One run, ONE stamp: the route publishes the sweep-built
|
||||
# payloads directly, so every event of this run carries the
|
||||
# same ``occurred_at`` the sweep computed for the whole run.
|
||||
assert payload["occurred_at"] == run_stamp
|
||||
|
||||
# Exactly one transition per instance: today's not-past-due
|
||||
# quest never arrives.
|
||||
with pytest.raises(asyncio.TimeoutError):
|
||||
await asyncio.wait_for(stream.queue.get(), timeout=0.3)
|
||||
|
||||
# The run recorded the watermark: the household-local date the
|
||||
# sweep ran for, under the well-known meta_state key.
|
||||
assert _watermark(ns.server) == seed.today_iso
|
||||
|
||||
|
||||
async def test_trigger_publishes_sweep_built_payloads_without_refetch(
|
||||
missed_app,
|
||||
) -> None:
|
||||
"""Publishing never re-fetches: a concurrently deleted instance
|
||||
cannot fail the trigger or lose its already-built missed event.
|
||||
|
||||
The core sweep commits the idempotency watermark BEFORE returning,
|
||||
so a publish loop that re-fetches each instance opens a loss
|
||||
window: an instance deleted inside that window (e.g. a concurrent
|
||||
regenerate) would raise ``ValueError`` uncaught — a bare 500 — and
|
||||
the missed event would be gone forever (the watermark already
|
||||
advanced, so no future sweep re-examines it). The route publishes
|
||||
the sweep-built ``(event_type, payload)`` pairs DIRECTLY (the
|
||||
api/scheduler.py shape), which this test exercises deterministically:
|
||||
the sweep is wrapped to delete one swept instance AFTER the sweep
|
||||
built its events and committed its watermark but BEFORE the route
|
||||
publishes, and the trigger must still answer 200 with ``fired: 2``
|
||||
and stream BOTH sweep-built payloads — the deleted instance's
|
||||
included, stamped with the run's ONE shared ``occurred_at``.
|
||||
"""
|
||||
ns = missed_app
|
||||
seed = ns.seed
|
||||
database = ns.server.app.state.db.database
|
||||
original_sweep = _core_sweep.run_missed_sweep
|
||||
|
||||
async def sweep_then_delete_stale(sweep_database, *, today):
|
||||
events = await original_sweep(sweep_database, today=today)
|
||||
# The concurrent deletion lands after the sweep built its
|
||||
# events and committed its watermark, but before the route
|
||||
# publishes: the row is gone while the built event survives.
|
||||
await sweep_database.execute(
|
||||
"DELETE FROM quest_instances WHERE id = ?",
|
||||
(seed.stale_instance.id,),
|
||||
)
|
||||
return events
|
||||
|
||||
_core_sweep.run_missed_sweep = sweep_then_delete_stale
|
||||
try:
|
||||
stream = _stream(ns.server.url("/api/v1/panel/events"))
|
||||
async with stream:
|
||||
response = await ns.client.post(
|
||||
"/api/v1/admin/missed-sweep", headers=_admin_headers()
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"fired": 2}
|
||||
|
||||
# The FIRST event is the deleted instance's — published
|
||||
# from the payload built before the deletion, never
|
||||
# re-fetched after it (a re-fetch would 500 here).
|
||||
event_type, payload = await stream.next_event()
|
||||
assert event_type == EVENT_QUEST_MISSED
|
||||
assert payload["instance_id"] == seed.stale_instance.id
|
||||
assert payload["child_name"] == "Ada"
|
||||
assert payload["quest_title"] == "Stale chore"
|
||||
assert payload["window"] == "morning"
|
||||
assert payload["due_date"] == seed.yesterday_iso
|
||||
assert payload["occurred_at"]
|
||||
run_stamp = payload["occurred_at"]
|
||||
|
||||
event_type, payload = await stream.next_event()
|
||||
assert event_type == EVENT_QUEST_MISSED
|
||||
assert payload["instance_id"] == seed.old_instance.id
|
||||
assert payload["occurred_at"] == run_stamp
|
||||
|
||||
with pytest.raises(asyncio.TimeoutError):
|
||||
await asyncio.wait_for(stream.queue.get(), timeout=0.3)
|
||||
finally:
|
||||
_core_sweep.run_missed_sweep = original_sweep
|
||||
|
||||
# The watermark advanced even though one instance was deleted
|
||||
# mid-flight — nothing was lost to the publish-time failure window.
|
||||
assert _watermark(ns.server) == seed.today_iso
|
||||
|
||||
|
||||
async def test_same_day_rerun_publishes_nothing(missed_app) -> None:
|
||||
"""A second trigger the same day publishes nothing (the watermark)."""
|
||||
ns = missed_app
|
||||
headers = _admin_headers()
|
||||
first = await ns.client.post("/api/v1/admin/missed-sweep", headers=headers)
|
||||
assert first.status_code == 200
|
||||
assert first.json() == {"fired": 2}
|
||||
|
||||
stream = _stream(ns.server.url("/api/v1/panel/events"))
|
||||
async with stream:
|
||||
second = await ns.client.post(
|
||||
"/api/v1/admin/missed-sweep", headers=headers
|
||||
)
|
||||
assert second.status_code == 200
|
||||
assert second.json() == {"fired": 0}
|
||||
with pytest.raises(asyncio.TimeoutError):
|
||||
# The watermark makes the rerun an empty no-op: the stream
|
||||
# stays silent.
|
||||
await asyncio.wait_for(stream.queue.get(), timeout=0.3)
|
||||
|
||||
|
||||
async def test_run_after_downtime_publishes_accumulated_window_once(
|
||||
missed_app,
|
||||
) -> None:
|
||||
"""After days of downtime the catch-up run sweeps the window once.
|
||||
|
||||
The first trigger sweeps yesterday's instances (watermark := today).
|
||||
``today`` then jumps two days: the accumulated window
|
||||
``[watermark, today)`` holds exactly the seeded TODAY-due instance
|
||||
(no completion events), which fires once — and a same-day rerun is
|
||||
silent again.
|
||||
"""
|
||||
ns = missed_app
|
||||
headers = _admin_headers()
|
||||
first = await ns.client.post("/api/v1/admin/missed-sweep", headers=headers)
|
||||
assert first.status_code == 200
|
||||
assert first.json() == {"fired": 2}
|
||||
|
||||
# Advance the pinned clock two days: the app is "down" over the
|
||||
# rollovers in between.
|
||||
routes_admin._local_now = lambda: ns.pinned_now + datetime.timedelta(
|
||||
days=2
|
||||
)
|
||||
later_today = (ns.today + datetime.timedelta(days=2)).isoformat()
|
||||
stream = _stream(ns.server.url("/api/v1/panel/events"))
|
||||
async with stream:
|
||||
catchup = await ns.client.post(
|
||||
"/api/v1/admin/missed-sweep", headers=headers
|
||||
)
|
||||
assert catchup.status_code == 200
|
||||
assert catchup.json() == {"fired": 1}
|
||||
|
||||
event_type, payload = await stream.next_event()
|
||||
assert event_type == EVENT_QUEST_MISSED
|
||||
assert payload["instance_id"] == ns.seed.recoverable_instance.id
|
||||
assert payload["child_id"] == ns.seed.ada.id
|
||||
assert payload["quest_title"] == "Recoverable chore"
|
||||
assert payload["window"] == "afternoon"
|
||||
assert payload["due_date"] == ns.seed.today_iso
|
||||
assert payload["due_time"] == "17:00"
|
||||
assert payload["occurred_at"]
|
||||
|
||||
with pytest.raises(asyncio.TimeoutError):
|
||||
await asyncio.wait_for(stream.queue.get(), timeout=0.3)
|
||||
|
||||
# The watermark moved to the new today...
|
||||
assert _watermark(ns.server) == later_today
|
||||
# ...so a further rerun publishes nothing.
|
||||
again = await ns.client.post("/api/v1/admin/missed-sweep", headers=headers)
|
||||
assert again.status_code == 200
|
||||
assert again.json() == {"fired": 0}
|
||||
|
||||
|
||||
# --- authentication: the router's ONE require_admin dependency ----------
|
||||
|
||||
|
||||
async def test_trigger_refuses_non_admin_and_absent_credentials(
|
||||
missed_app,
|
||||
) -> None:
|
||||
"""A non-admin JWT is 403, an absent credential is 401; none sweep."""
|
||||
ns = missed_app
|
||||
non_admin = make_token(groups=("some-other-group",))
|
||||
forbidden = await ns.client.post(
|
||||
"/api/v1/admin/missed-sweep",
|
||||
headers={"Authorization": f"Bearer {non_admin}"},
|
||||
)
|
||||
assert forbidden.status_code == 403
|
||||
assert forbidden.json()["detail"] == (
|
||||
"Admin access requires membership in the nestquest-admins group"
|
||||
)
|
||||
absent = await ns.client.post("/api/v1/admin/missed-sweep")
|
||||
assert absent.status_code == 401
|
||||
|
||||
# Neither refusal ran the sweep: the watermark is still unset, and
|
||||
# the next admin trigger fires the full seeded set.
|
||||
assert _watermark(ns.server) is None
|
||||
after = await ns.client.post(
|
||||
"/api/v1/admin/missed-sweep", headers=_admin_headers()
|
||||
)
|
||||
assert after.status_code == 200
|
||||
assert after.json() == {"fired": 2}
|
||||
|
||||
|
||||
# --- the daily scheduler -------------------------------------------------
|
||||
|
||||
|
||||
class _RecordingPublisher:
|
||||
"""A transition-publisher stub recording every published event."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.published: list[tuple[str, dict]] = []
|
||||
|
||||
def publish(self, event_type: str, payload: dict) -> None:
|
||||
self.published.append((event_type, payload))
|
||||
|
||||
|
||||
async def test_scheduler_runs_sweep_at_rollover_and_cancels_cleanly(
|
||||
temp_db_path: str,
|
||||
) -> None:
|
||||
"""The scheduler sweeps at the configured rollover and stops clean.
|
||||
|
||||
The injectable clock starts at 10:00 and the settings pin the
|
||||
rollover to 10:05, so the first computed delay is five minutes;
|
||||
the injectable sleep fast-forwards the clock, the sweep runs for
|
||||
the (now advanced) clock's date, and the seeded past-due open
|
||||
instance is published on the recording publisher exactly once.
|
||||
The next cycle schedules for the NEXT day's rollover (23h55m).
|
||||
``stop`` cancels the loop task and the scheduler reports stopped.
|
||||
"""
|
||||
database = _core_db.NestQuestDatabase(_executor)
|
||||
await database.open(temp_db_path)
|
||||
try:
|
||||
await _core_migrations.apply_migrations(database)
|
||||
now_local = datetime.datetime.now().astimezone()
|
||||
start = now_local.replace(
|
||||
hour=10, minute=0, second=0, microsecond=0
|
||||
)
|
||||
ada = await _children.create_child(database, "Ada", sort_order=0)
|
||||
yesterday = (start.date() - datetime.timedelta(days=1)).isoformat()
|
||||
await _quest_definitions.create_quest_definition(
|
||||
database,
|
||||
"Stale chore",
|
||||
_recurrence.ScheduleRule.from_dict(
|
||||
{
|
||||
"rule_type": "daily",
|
||||
"start_date": yesterday,
|
||||
"end_date": yesterday,
|
||||
}
|
||||
),
|
||||
[ada.id],
|
||||
[("morning", "09:00")],
|
||||
)
|
||||
await _materialize.materialize(
|
||||
database, yesterday, yesterday, today=start.date() - datetime.timedelta(days=1)
|
||||
)
|
||||
await _settings_store.update_settings(
|
||||
database, {"day_rollover_time": "10:05"}
|
||||
)
|
||||
|
||||
clock_state = {"now": start}
|
||||
delays: list[float] = []
|
||||
|
||||
def clock() -> datetime.datetime:
|
||||
return clock_state["now"]
|
||||
|
||||
async def sleep_stub(seconds: float) -> None:
|
||||
delays.append(seconds)
|
||||
if len(delays) == 1:
|
||||
# Fast-forward the clock across the rollover wait ONCE,
|
||||
# so the first sweep runs at the rollover instant; the
|
||||
# clock then stays frozen at that instant, so the
|
||||
# follow-up cycles sweep the same (already-watermarked)
|
||||
# date and the assertions read a stable state.
|
||||
clock_state["now"] = clock_state["now"] + datetime.timedelta(
|
||||
seconds=seconds
|
||||
)
|
||||
await asyncio.sleep(0)
|
||||
|
||||
publisher = _RecordingPublisher()
|
||||
scheduler = MissedSweepScheduler(
|
||||
database, publisher=publisher, clock=clock, sleep=sleep_stub
|
||||
)
|
||||
scheduler.start()
|
||||
try:
|
||||
deadline = time.monotonic() + READ_TIMEOUT
|
||||
while not publisher.published:
|
||||
if time.monotonic() > deadline:
|
||||
pytest.fail("scheduler never published a missed event")
|
||||
await asyncio.sleep(0.01)
|
||||
# The second cycle's scheduling decision may lag the first
|
||||
# publication by a few awaits; wait for it the same way.
|
||||
while len(delays) < 2:
|
||||
if time.monotonic() > deadline:
|
||||
pytest.fail("scheduler never scheduled a second cycle")
|
||||
await asyncio.sleep(0.01)
|
||||
|
||||
event_type, payload = publisher.published[0]
|
||||
assert event_type == "nestquest_quest_missed"
|
||||
assert payload["child_id"] == ada.id
|
||||
assert payload["quest_title"] == "Stale chore"
|
||||
assert payload["window"] == "morning"
|
||||
assert payload["due_date"] == yesterday
|
||||
assert payload["due_time"] == "09:00"
|
||||
assert payload["occurred_at"]
|
||||
|
||||
# The rollover delay was honored exactly...
|
||||
assert delays[0] == pytest.approx(300.0)
|
||||
# ...and the next cycle targets the NEXT day's rollover
|
||||
# (the just-passed instant schedules for tomorrow).
|
||||
assert delays[1] == pytest.approx(86400.0)
|
||||
# The watermark pins the run to the sweep-day's date.
|
||||
watermark = await _dao_meta.MetaStateDao(database).get(
|
||||
SWEEP_WATERMARK_KEY
|
||||
)
|
||||
assert watermark == start.date().isoformat()
|
||||
finally:
|
||||
await scheduler.stop()
|
||||
assert not scheduler.running
|
||||
# Exactly one announcement: the watermark makes every further
|
||||
# cycle an empty no-op.
|
||||
assert len(publisher.published) == 1
|
||||
finally:
|
||||
await database.close()
|
||||
Reference in New Issue
Block a user