Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1add535628 |
@@ -1,27 +0,0 @@
|
||||
# Keep the image build context minimal: only api/ and the bundled
|
||||
# custom_components/nestquest/core/ are copied (see the Dockerfile);
|
||||
# everything else — VCS state, virtualenvs, tests, the frontend, and
|
||||
# any local secrets or database files — never enters the build.
|
||||
.git
|
||||
.gitignore
|
||||
.venv
|
||||
**/__pycache__
|
||||
**/*.py[cod]
|
||||
.pytest_cache
|
||||
**/*.db
|
||||
**/*.db-wal
|
||||
**/*.db-shm
|
||||
# Local configuration — never baked into an image.
|
||||
.env
|
||||
**/.env
|
||||
deploy
|
||||
design
|
||||
docs
|
||||
frontend
|
||||
orchestration_tests
|
||||
scripts
|
||||
templates
|
||||
tests
|
||||
uv.lock
|
||||
hacs.json
|
||||
*.md
|
||||
@@ -5,12 +5,3 @@ __pycache__/
|
||||
.pytest_cache/
|
||||
*.metadata.json
|
||||
frontend/node_modules/
|
||||
|
||||
# Feature 17 (API deployment): local secret config and runtime data
|
||||
# are never committed — compose reads deploy/.env; SQLite/WAL sidecar
|
||||
# files are container/volume artifacts.
|
||||
.env
|
||||
**/.env
|
||||
*.db
|
||||
*.db-wal
|
||||
*.db-shm
|
||||
|
||||
-47
@@ -1,47 +0,0 @@
|
||||
# Container image for the NestQuest API service (Feature 17).
|
||||
#
|
||||
# Build context is the REPOSITORY ROOT (compose sets it), because the
|
||||
# API loads the bundled domain layer from
|
||||
# custom_components/nestquest/core/ by file location
|
||||
# (api/nestquest_core.py): the image must carry both api/ and the
|
||||
# bundled core at their relative positions. Everything else in the
|
||||
# repository is excluded via .dockerignore.
|
||||
#
|
||||
# Multi-stage so no pip/build toolchain survives into the runtime:
|
||||
# the builder installs the deployable's own manifest
|
||||
# (api/requirements.txt) into a staging prefix and the runtime copies
|
||||
# only the installed packages. The service runs as a non-root user;
|
||||
# the SQLite database lives on the mounted volume at
|
||||
# /var/lib/nestquest (compose supplies NESTQUEST_DB_PATH pointing
|
||||
# there — configuration is environment-supplied, never baked in).
|
||||
FROM python:3.12-slim AS builder
|
||||
|
||||
COPY api/requirements.txt /tmp/requirements.txt
|
||||
RUN pip install --no-cache-dir --prefix=/install -r /tmp/requirements.txt
|
||||
|
||||
FROM python:3.12-slim
|
||||
|
||||
# The staged dependency install from the builder (no pip, no wheels).
|
||||
COPY --from=builder /install /usr/local
|
||||
|
||||
# /app is the repository-root equivalent: api/ resolves imports as
|
||||
# ``api.*`` and nestquest_core.py finds the bundled core one level up.
|
||||
WORKDIR /app
|
||||
COPY api/ ./api/
|
||||
COPY custom_components/nestquest/core/ ./custom_components/nestquest/core/
|
||||
|
||||
# Non-root runtime user; owns the database volume mount point.
|
||||
RUN useradd --system --create-home --home-dir /home/nestquest nestquest \
|
||||
&& mkdir -p /var/lib/nestquest \
|
||||
&& chown nestquest:nestquest /var/lib/nestquest
|
||||
USER nestquest
|
||||
|
||||
ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1
|
||||
|
||||
# The database path and the service token / OIDC settings are supplied
|
||||
# by the environment (compose env_file), never committed into the image.
|
||||
ENV NESTQUEST_DB_PATH=/var/lib/nestquest/nestquest.db
|
||||
VOLUME /var/lib/nestquest
|
||||
|
||||
EXPOSE 8080
|
||||
CMD ["uvicorn", "--factory", "api.app:create_app", "--host", "0.0.0.0", "--port", "8080"]
|
||||
@@ -1,32 +0,0 @@
|
||||
# NestQuest API
|
||||
|
||||
Standalone FastAPI service owning the NestQuest database (Feature 16).
|
||||
Served with ``uvicorn --factory api.app:create_app``; configuration is
|
||||
environment-supplied (``api/config.py``) and every variable is required.
|
||||
|
||||
## Runtime dependencies
|
||||
|
||||
``api/requirements.txt`` is the API deployable's OWN dependency
|
||||
manifest (fastapi, uvicorn, pyjwt[crypto], httpx) — deliberately kept
|
||||
out of the integration's ``pyproject.toml`` dev group, which exists only
|
||||
so the monorepo test venv can run the API tests.
|
||||
|
||||
## Container deployment (scaffold)
|
||||
|
||||
The image is built from the repository ROOT context (the API loads the
|
||||
bundled domain core from ``custom_components/nestquest/core/`` by file
|
||||
location, so both trees must be in the build context):
|
||||
|
||||
docker build -t nestquest-api:latest .
|
||||
|
||||
Run it with compose — the SQLite file lives on the named volume
|
||||
``nestquest-db`` so recreating the container keeps household data:
|
||||
|
||||
cp deploy/.env.example deploy/.env # then fill in real values
|
||||
docker compose -f deploy/compose.yaml up -d --build
|
||||
|
||||
``deploy/.env`` is git-ignored and supplies the secrets (panel service
|
||||
token, Authentik OIDC settings); ``NESTQUEST_DB_PATH`` is fixed to
|
||||
``/var/lib/nestquest/nestquest.db`` by compose. The port is bound to
|
||||
localhost only — TLS/reverse proxy, Authentik, network separation,
|
||||
backup and restore are later Feature 17 tasks.
|
||||
@@ -1,15 +0,0 @@
|
||||
# Runtime dependency manifest for the NestQuest API deployable.
|
||||
#
|
||||
# The monorepo's pyproject.toml carries fastapi/uvicorn/pyjwt/httpx in
|
||||
# the INTEGRATION's dev group only (so the test venv can exercise the
|
||||
# API); they are deliberately NOT integration runtime dependencies.
|
||||
# Per that manifest's own comment, the API deployable declares its own
|
||||
# runtime dependencies HERE — this file is what the container image
|
||||
# installs (see the Dockerfile at the repository root).
|
||||
#
|
||||
# Compatible ranges mirroring the dev group's floors; the API is a
|
||||
# server, so these are the only packages its runtime needs.
|
||||
fastapi>=0.115
|
||||
uvicorn>=0.30
|
||||
pyjwt[crypto]>=2.9
|
||||
httpx>=0.28
|
||||
@@ -41,4 +41,38 @@ class MetaStateDao:
|
||||
f"INSERT INTO {_TABLE} (key, value) VALUES (?, ?) "
|
||||
"ON CONFLICT (key) DO UPDATE SET value = excluded.value",
|
||||
(key, value),
|
||||
)
|
||||
)
|
||||
|
||||
async def claim(self, key: str, value: str) -> bool:
|
||||
"""Atomically set ``key`` to ``value`` unless it already is.
|
||||
|
||||
The single upsert statement carries a ``WHERE`` guard on the
|
||||
conflict path (``t.value IS NOT excluded.value``), so the
|
||||
read-compare-write is ONE SQLite statement — atomic at the
|
||||
DATABASE level, not just within one process: two connections
|
||||
(two processes sharing one SQLite file) cannot both observe
|
||||
"not yet set" and both write, because the second writer's
|
||||
statement re-evaluates the guard against the first writer's
|
||||
committed value and matches zero rows.
|
||||
|
||||
Returns True only when THIS call changed the row: the first
|
||||
claimant wins and every later claimant of the same value
|
||||
returns False. A DIFFERENT value always wins (an advance, e.g.
|
||||
the sweep watermark moving to a new date) and re-arms the
|
||||
same-value loser check for the new value.
|
||||
|
||||
Runs under the shared connection lock and inside one
|
||||
transaction, like :meth:`set`; ``cursor.rowcount`` (1 for a
|
||||
matched write, 0 for a guarded no-op) is captured by the
|
||||
wrapper on the executor thread, so no sqlite3 object crosses
|
||||
back to the event loop.
|
||||
"""
|
||||
async with _connection_lock(self._database):
|
||||
async with self._database.transaction():
|
||||
result = await self._database.execute(
|
||||
f"INSERT INTO {_TABLE} (key, value) VALUES (?, ?) "
|
||||
"ON CONFLICT (key) DO UPDATE SET value = excluded.value "
|
||||
f"WHERE {_TABLE}.value IS NOT excluded.value",
|
||||
(key, value),
|
||||
)
|
||||
return result.rowcount == 1
|
||||
|
||||
@@ -26,12 +26,17 @@ 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
|
||||
The whole watermark read → query → build 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.
|
||||
a rollover sweep of the SAME process queue instead of interleaving.
|
||||
The announce decision itself is the DATABASE-level claim
|
||||
(:meth:`MetaStateDao.claim`), so two PROCESSES sharing one SQLite file
|
||||
(the integration's rollover sweep and the API service's scheduler)
|
||||
cannot both announce the same missed batch either: both may read the
|
||||
old watermark and build the events, but the single conditional upsert
|
||||
lets exactly one of them win the claim — the loser returns [] and
|
||||
announces nothing.
|
||||
|
||||
The caller supplies ``today`` as a plain :class:`datetime.date`: the
|
||||
household-local calendar date, read from ``hass.config.time_zone`` by
|
||||
@@ -60,11 +65,11 @@ LOGGER = logging.getLogger(__name__)
|
||||
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.
|
||||
#: the settings store's pattern: the watermark read → query → build
|
||||
#: span is one critical section, so an overlapping startup sweep and
|
||||
#: rollover sweep of the SAME process queue instead of interleaving.
|
||||
#: Same-process callers are fully serialised by it; the cross-process
|
||||
#: case is closed by the database-level claim, not by this lock.
|
||||
_SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
|
||||
|
||||
|
||||
@@ -99,15 +104,24 @@ async def run_missed_sweep(
|
||||
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
|
||||
The watermark read → query → build span runs inside the
|
||||
per-database sweep lock, so overlapping callers of ONE process
|
||||
queue instead of interleaving. The announce decision is the
|
||||
DATABASE-level claim (:meth:`MetaStateDao.claim` — a single
|
||||
conditional upsert on the watermark, atomic across processes):
|
||||
every caller builds the events optimistically, then only the
|
||||
caller whose claim actually advanced the watermark returns them;
|
||||
a loser (another process already announced for ``today``) returns
|
||||
[] 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.
|
||||
|
||||
Failure ordering: the watermark is advanced ONLY by the claim,
|
||||
AFTER the events are fully built — a failure before the claim
|
||||
leaves the watermark unadvanced, so a later run re-sweeps the same
|
||||
window and nothing is lost.
|
||||
"""
|
||||
today_iso = today.isoformat()
|
||||
dao = MetaStateDao(database)
|
||||
@@ -134,10 +148,23 @@ async def run_missed_sweep(
|
||||
),
|
||||
)
|
||||
)
|
||||
await dao.set(SWEEP_WATERMARK_KEY, today_iso)
|
||||
# The database-level claim gates the announcement: only the
|
||||
# caller that actually advanced the watermark may publish the
|
||||
# built events, so two processes racing one SQLite file cannot
|
||||
# both announce the same missed batch. A same-value loser's
|
||||
# claim matches zero rows and this run announces nothing.
|
||||
claimed = await dao.claim(SWEEP_WATERMARK_KEY, today_iso)
|
||||
if not claimed:
|
||||
LOGGER.info(
|
||||
"NestQuest missed sweep for %s lost the watermark claim; "
|
||||
"announcing nothing",
|
||||
today_iso,
|
||||
)
|
||||
return []
|
||||
LOGGER.info(
|
||||
"NestQuest missed sweep for %s built %d event(s)",
|
||||
today_iso,
|
||||
len(events),
|
||||
)
|
||||
return events
|
||||
|
||||
|
||||
@@ -1,17 +0,0 @@
|
||||
# Example environment for the NestQuest API container deployment.
|
||||
#
|
||||
# Copy to ``deploy/.env`` (git-ignored) and fill in REAL values there;
|
||||
# this example carries placeholder names only and is never a source of
|
||||
# secrets. Compose loads ``deploy/.env`` into the api service; the
|
||||
# API fails fast on startup when any of these is unset or empty
|
||||
# (api/config.py — there are no silent defaults).
|
||||
|
||||
# The static panel-plane service token the HA integration presents as
|
||||
# a Bearer credential on /api/v1/panel/* routes. Generate one, e.g.:
|
||||
# python -c "import secrets; print(secrets.token_urlsafe(32))"
|
||||
NESTQUEST_PANEL_TOKEN=
|
||||
|
||||
# The Authentik OIDC settings the admin plane verifies JWTs against.
|
||||
NESTQUEST_OIDC_ISSUER=
|
||||
NESTQUEST_OIDC_AUDIENCE=
|
||||
NESTQUEST_OIDC_JWKS_URL=
|
||||
@@ -1,41 +0,0 @@
|
||||
# Compose deployment scaffold for the NestQuest API service (Feature
|
||||
# 17). The SQLite database lives on the NAMED VOLUME ``nestquest-db``
|
||||
# mounted at /var/lib/nestquest — recreating the container keeps the
|
||||
# household data (``docker compose down`` leaves named volumes alone).
|
||||
#
|
||||
# Configuration is supplied by the environment: the fixed database
|
||||
# path here, and the secret values (panel service token, Authentik
|
||||
# OIDC settings) from ``deploy/.env`` — a git-ignored file you create
|
||||
# from ``deploy/.env.example`` (placeholder names only; never commit
|
||||
# real values). TLS/reverse proxy, Authentik wiring and network
|
||||
# separation are later Feature 17 tasks; the port is bound to
|
||||
# localhost only until those land.
|
||||
services:
|
||||
api:
|
||||
build:
|
||||
context: ..
|
||||
dockerfile: Dockerfile
|
||||
image: nestquest-api:latest
|
||||
init: true
|
||||
restart: unless-stopped
|
||||
ports:
|
||||
- "127.0.0.1:8080:8080"
|
||||
environment:
|
||||
# The SQLite file inside the named volume; fixed, not a secret.
|
||||
NESTQUEST_DB_PATH: /var/lib/nestquest/nestquest.db
|
||||
env_file:
|
||||
# Secret values (NESTQUEST_PANEL_TOKEN, NESTQUEST_OIDC_*); absent
|
||||
# until you create deploy/.env from the example.
|
||||
- path: .env
|
||||
required: false
|
||||
volumes:
|
||||
- nestquest-db:/var/lib/nestquest
|
||||
healthcheck:
|
||||
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8080/health')"]
|
||||
interval: 10s
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
start_period: 15s
|
||||
|
||||
volumes:
|
||||
nestquest-db:
|
||||
@@ -0,0 +1,295 @@
|
||||
"""Cross-process safety of the missed-sweep watermark claim.
|
||||
|
||||
Feature 11's sweep guards its announcement with the watermark; the
|
||||
original guard was a read-then-set inside a per-PROCESS asyncio lock,
|
||||
so two PROCESSES sharing one SQLite file (the HA integration's
|
||||
rollover sweep and the API service's scheduler) could both read the
|
||||
old watermark and both announce the same missed batch. These tests
|
||||
pin the fix:
|
||||
|
||||
- :meth:`MetaStateDao.claim` is the atomic, database-level primitive:
|
||||
one conditional upsert whose ``rowcount`` says whether THIS caller
|
||||
changed the row. The first claimant of a value wins; a second
|
||||
claimant of the same value — even on an INDEPENDENT connection to
|
||||
the same file, the shape two processes produce — loses.
|
||||
- :func:`run_missed_sweep` gates the announcement on that claim: two
|
||||
independent ``NestQuestDatabase`` wrappers (two processes' shape)
|
||||
racing one seeded database yields exactly one announcer.
|
||||
- The documented sweep semantics are unchanged: a same-night rerun
|
||||
announces nothing, and a run after days of downtime announces the
|
||||
accumulated ``[watermark, today)`` window exactly once.
|
||||
|
||||
Everything here runs against the bundled HA-free core copy
|
||||
(``custom_components.nestquest.core``) with a plain
|
||||
``asyncio.to_thread`` executor — no Home Assistant, no API server.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import datetime
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from custom_components.nestquest.core.dao_instances import (
|
||||
QuestInstancesDao,
|
||||
)
|
||||
from custom_components.nestquest.core.dao_meta import MetaStateDao
|
||||
from custom_components.nestquest.core.db import NestQuestDatabase
|
||||
from custom_components.nestquest.core.migrations import apply_migrations
|
||||
from custom_components.nestquest.core.children import create_child
|
||||
from custom_components.nestquest.core.quest_definitions import (
|
||||
create_quest_definition,
|
||||
)
|
||||
from custom_components.nestquest.core.materialize import materialize
|
||||
from custom_components.nestquest.core.recurrence import ScheduleRule
|
||||
from custom_components.nestquest.core.sweep import (
|
||||
SWEEP_WATERMARK_KEY,
|
||||
run_missed_sweep,
|
||||
)
|
||||
|
||||
#: One missed event per unswept past-due instance — the sweep's only
|
||||
#: event type.
|
||||
EVENT_QUEST_MISSED = "nestquest_quest_missed"
|
||||
|
||||
|
||||
async def _executor(fn, *args, **kwargs):
|
||||
"""Run a synchronous callable on a worker thread and await it.
|
||||
|
||||
The :class:`NestQuestDatabase` executor contract (the shape of
|
||||
``hass.async_add_executor_job``), so sqlite3 calls never block the
|
||||
event loop.
|
||||
"""
|
||||
return await asyncio.to_thread(fn, *args, **kwargs)
|
||||
|
||||
|
||||
async def _open_database(db_path: str) -> NestQuestDatabase:
|
||||
"""Open (without migrating) one independent connection wrapper."""
|
||||
database = NestQuestDatabase(_executor)
|
||||
await database.open(db_path)
|
||||
return database
|
||||
|
||||
|
||||
async def _seed_missed_household(
|
||||
db_path: str, today: datetime.date
|
||||
) -> SimpleNamespace:
|
||||
"""Seed a household with two past-due open instances on ``db_path``.
|
||||
|
||||
One child, two quests whose ONLY instances are due YESTERDAY
|
||||
(materialized with ``today`` pinned to yesterday, ``end_date``
|
||||
closing the rules so the later today-walk cannot duplicate them).
|
||||
Seeding uses its OWN connection, opened and closed here, so the
|
||||
connections the tests race with start fresh.
|
||||
"""
|
||||
database = await _open_database(db_path)
|
||||
try:
|
||||
await apply_migrations(database)
|
||||
today_iso = today.isoformat()
|
||||
yesterday = today - datetime.timedelta(days=1)
|
||||
yesterday_iso = yesterday.isoformat()
|
||||
rule = ScheduleRule.from_dict(
|
||||
{
|
||||
"rule_type": "daily",
|
||||
"start_date": yesterday_iso,
|
||||
"end_date": yesterday_iso,
|
||||
}
|
||||
)
|
||||
ada = await create_child(database, "Ada", sort_order=0)
|
||||
await create_quest_definition(
|
||||
database, "Stale chore", rule, [ada.id], [("morning", "09:00")]
|
||||
)
|
||||
await create_quest_definition(
|
||||
database, "Old chore", rule, [ada.id], [("evening", "18:30")]
|
||||
)
|
||||
await materialize(database, yesterday_iso, yesterday_iso, today=yesterday)
|
||||
# A quest due TODAY (created after the yesterday walk so it has
|
||||
# no past-due instance): the post-downtime catch-up run's
|
||||
# target — inside [watermark=today, later_today) but not
|
||||
# inside the day-one sweep's window.
|
||||
await create_quest_definition(
|
||||
database,
|
||||
"Recoverable chore",
|
||||
ScheduleRule.from_dict(
|
||||
{"rule_type": "daily", "start_date": today_iso}
|
||||
),
|
||||
[ada.id],
|
||||
[("afternoon", "17:00")],
|
||||
)
|
||||
await materialize(database, today_iso, today_iso, today=today)
|
||||
instances = QuestInstancesDao(database)
|
||||
rows = await instances.list_by_child_and_date(ada.id, yesterday_iso)
|
||||
assert len(rows) == 2
|
||||
today_rows = await instances.list_by_child_and_date(
|
||||
ada.id, today_iso
|
||||
)
|
||||
assert len(today_rows) == 1
|
||||
return SimpleNamespace(
|
||||
child=ada,
|
||||
instances=rows,
|
||||
recoverable_instance=today_rows[0],
|
||||
yesterday_iso=yesterday_iso,
|
||||
today_iso=today_iso,
|
||||
)
|
||||
finally:
|
||||
await database.close()
|
||||
|
||||
|
||||
# --- the claim primitive: atomic at the database level ------------------
|
||||
|
||||
|
||||
async def test_claim_first_caller_wins_and_second_same_value_loses(
|
||||
temp_db_path: str,
|
||||
) -> None:
|
||||
"""claim() is a conditional upsert: first True, same-value second False.
|
||||
|
||||
A different value always wins (an advance re-arms the check), and
|
||||
the stored value follows the winner.
|
||||
"""
|
||||
database = await _open_database(temp_db_path)
|
||||
try:
|
||||
await apply_migrations(database)
|
||||
dao = MetaStateDao(database)
|
||||
assert await dao.get(SWEEP_WATERMARK_KEY) is None
|
||||
|
||||
assert await dao.claim(SWEEP_WATERMARK_KEY, "2026-09-01") is True
|
||||
# Same value again: the row is already there — nothing changed.
|
||||
assert await dao.claim(SWEEP_WATERMARK_KEY, "2026-09-01") is False
|
||||
assert await dao.get(SWEEP_WATERMARK_KEY) == "2026-09-01"
|
||||
# A DIFFERENT value is an advance, not a re-claim: it wins.
|
||||
assert await dao.claim(SWEEP_WATERMARK_KEY, "2026-09-02") is True
|
||||
assert await dao.get(SWEEP_WATERMARK_KEY) == "2026-09-02"
|
||||
assert await dao.claim(SWEEP_WATERMARK_KEY, "2026-09-02") is False
|
||||
finally:
|
||||
await database.close()
|
||||
|
||||
|
||||
async def test_claim_is_atomic_across_independent_connections(
|
||||
temp_db_path: str,
|
||||
) -> None:
|
||||
"""Two connections to one file cannot both claim the same value.
|
||||
|
||||
This is the two-process shape: independent sqlite3 connections (one
|
||||
per ``NestQuestDatabase`` wrapper), racing the SAME key on the SAME
|
||||
event loop with no shared lock. The single conditional upsert is
|
||||
serialized by SQLite itself, so exactly one caller's rowcount is 1.
|
||||
"""
|
||||
first = await _open_database(temp_db_path)
|
||||
second = await _open_database(temp_db_path)
|
||||
try:
|
||||
await apply_migrations(first)
|
||||
won = await asyncio.gather(
|
||||
MetaStateDao(first).claim(SWEEP_WATERMARK_KEY, "2026-09-01"),
|
||||
MetaStateDao(second).claim(SWEEP_WATERMARK_KEY, "2026-09-01"),
|
||||
)
|
||||
assert sorted(won) == [False, True]
|
||||
# One stored row, holding the winner's value.
|
||||
assert await MetaStateDao(first).get(SWEEP_WATERMARK_KEY) == (
|
||||
"2026-09-01"
|
||||
)
|
||||
assert await MetaStateDao(second).get(SWEEP_WATERMARK_KEY) == (
|
||||
"2026-09-01"
|
||||
)
|
||||
finally:
|
||||
await first.close()
|
||||
await second.close()
|
||||
|
||||
|
||||
# --- the sweep: the announcement is gated by the atomic claim -----------
|
||||
|
||||
|
||||
async def test_two_process_shape_race_exactly_one_announces(
|
||||
temp_db_path: str,
|
||||
) -> None:
|
||||
"""Two independent connections racing the sweep: ONE announcer.
|
||||
|
||||
Each wrapper carries its own connection AND its own per-process
|
||||
sweep lock, so nothing in-process serialises the two runs — the
|
||||
only guard left is the database-level claim. Both read the unset
|
||||
watermark and both build the events, but exactly one claim wins:
|
||||
one run returns the missed events, the loser returns [].
|
||||
"""
|
||||
today = datetime.datetime.now().astimezone().date()
|
||||
seed = await _seed_missed_household(temp_db_path, today)
|
||||
first = await _open_database(temp_db_path)
|
||||
second = await _open_database(temp_db_path)
|
||||
try:
|
||||
results = await asyncio.gather(
|
||||
run_missed_sweep(first, today=today),
|
||||
run_missed_sweep(second, today=today),
|
||||
)
|
||||
announcers = [r for r in results if r]
|
||||
assert len(announcers) == 1
|
||||
event_type, payload = announcers[0][0]
|
||||
assert event_type == EVENT_QUEST_MISSED
|
||||
assert payload["due_date"] == seed.yesterday_iso
|
||||
# Both instances of the seeded window, one event each.
|
||||
assert len(announcers[0]) == 2
|
||||
losers = [r for r in results if not r]
|
||||
assert losers == [[]]
|
||||
# The watermark holds the run's date for whoever looks next.
|
||||
assert (
|
||||
await MetaStateDao(first).get(SWEEP_WATERMARK_KEY)
|
||||
== seed.today_iso
|
||||
)
|
||||
finally:
|
||||
await first.close()
|
||||
await second.close()
|
||||
|
||||
|
||||
async def test_same_night_rerun_announces_nothing(temp_db_path: str) -> None:
|
||||
"""A second sweep for the same date is an empty no-op (the watermark)."""
|
||||
today = datetime.datetime.now().astimezone().date()
|
||||
await _seed_missed_household(temp_db_path, today)
|
||||
database = await _open_database(temp_db_path)
|
||||
try:
|
||||
first = await run_missed_sweep(database, today=today)
|
||||
assert len(first) == 2
|
||||
second = await run_missed_sweep(database, today=today)
|
||||
assert second == []
|
||||
finally:
|
||||
await database.close()
|
||||
|
||||
|
||||
async def test_post_downtime_run_announces_accumulated_window_once(
|
||||
temp_db_path: str,
|
||||
) -> None:
|
||||
"""After days of downtime the catch-up run sweeps the window once.
|
||||
|
||||
The first run sweeps yesterday's instances (watermark := today);
|
||||
``today`` then jumps two days and the catch-up run — on a
|
||||
DIFFERENT connection, the two-process shape — announces the
|
||||
accumulated ``[watermark, later_today)`` window exactly once, and
|
||||
a rerun on the original connection is silent again.
|
||||
"""
|
||||
today = datetime.datetime.now().astimezone().date()
|
||||
later_today = today + datetime.timedelta(days=2)
|
||||
seed = await _seed_missed_household(temp_db_path, today)
|
||||
first = await _open_database(temp_db_path)
|
||||
second = await _open_database(temp_db_path)
|
||||
try:
|
||||
# Day one: yesterday's two instances fire once.
|
||||
day_one = await run_missed_sweep(first, today=today)
|
||||
assert len(day_one) == 2
|
||||
assert all(
|
||||
payload["due_date"] == seed.yesterday_iso
|
||||
for _, payload in day_one
|
||||
)
|
||||
|
||||
# The downtime catch-up: the window is [watermark, later_today).
|
||||
catchup = await run_missed_sweep(second, today=later_today)
|
||||
assert len(catchup) == 1
|
||||
event_type, payload = catchup[0]
|
||||
assert event_type == EVENT_QUEST_MISSED
|
||||
assert payload["due_date"] == seed.today_iso
|
||||
assert payload["instance_id"] == seed.recoverable_instance.id
|
||||
assert payload["child_id"] == seed.child.id
|
||||
|
||||
# A further rerun announces nothing.
|
||||
assert await run_missed_sweep(first, today=later_today) == []
|
||||
assert (
|
||||
await MetaStateDao(first).get(SWEEP_WATERMARK_KEY)
|
||||
== later_today.isoformat()
|
||||
)
|
||||
finally:
|
||||
await first.close()
|
||||
await second.close()
|
||||
Reference in New Issue
Block a user