Re-plan the missed sweep when the panel adopts a time zone
The scheduler loaded settings once per cycle and slept on them until the old rollover, so a zone adopted from Home Assistant after start-up left a UTC host sweeping at UTC midnight (20:00 in New York), marking quests missed hours early. `MissedSweepScheduler.settings_changed` now wakes the sleeping loop when the stored zone or rollover time differs from the one it planned with; the panel's household clock calls it after adopting a reported zone. Before sweeping, the loop re-reads the settings and re-plans instead if they no longer match its plan. Tests: the scheduler re-plans for New York midnight when a panel request adopts the zone mid-sleep, and skips a stale sweep when the zone changed without a wake-up. The admin-precedence and invalid-header tests now prove adoption first, and the SSE client test asserts the timezone header rides beside the Authorization header. Task: 6351b36d-0eef-4e7d-bf5a-96b754cbc883 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
+4
-1
@@ -198,7 +198,9 @@ async def _household_clock(
|
|||||||
Home Assistant's ``time_zone`` in :data:`PANEL_TIMEZONE_HEADER`;
|
Home Assistant's ``time_zone`` in :data:`PANEL_TIMEZONE_HEADER`;
|
||||||
until the admin chooses a zone it is adopted and persisted
|
until the admin chooses a zone it is adopted and persisted
|
||||||
(:func:`nestquest_core.settings_store.adopt_reported_timezone`), so
|
(:func:`nestquest_core.settings_store.adopt_reported_timezone`), so
|
||||||
the scheduler and the admin plane follow it too. An invalid
|
the admin plane follows it too, and the already-sleeping missed-sweep
|
||||||
|
scheduler is told (:meth:`MissedSweepScheduler.settings_changed`) so
|
||||||
|
it re-plans for the household rollover. An invalid
|
||||||
reported zone is logged and ignored — the panel keeps serving on
|
reported zone is logged and ignored — the panel keeps serving on
|
||||||
the stored settings.
|
the stored settings.
|
||||||
"""
|
"""
|
||||||
@@ -208,6 +210,7 @@ async def _household_clock(
|
|||||||
settings = await core_settings_store.adopt_reported_timezone(
|
settings = await core_settings_store.adopt_reported_timezone(
|
||||||
database, reported
|
database, reported
|
||||||
)
|
)
|
||||||
|
request.app.state.sweep_scheduler.settings_changed(settings)
|
||||||
except ValueError:
|
except ValueError:
|
||||||
LOGGER.warning(
|
LOGGER.warning(
|
||||||
"Ignoring invalid %s header value %r",
|
"Ignoring invalid %s header value %r",
|
||||||
|
|||||||
+53
-1
@@ -13,6 +13,14 @@ local date when unset) and publishes each returned
|
|||||||
``nestquest_quest_missed`` event on the app's ONE transition publisher
|
``nestquest_quest_missed`` event on the app's ONE transition publisher
|
||||||
(the same SSE stream every route publishes through).
|
(the same SSE stream every route publishes through).
|
||||||
|
|
||||||
|
A household zone adopted from Home Assistant while the loop is asleep
|
||||||
|
must not leave it waiting for the OLD zone's rollover (a UTC host would
|
||||||
|
otherwise sweep at UTC midnight — 20:00 in New York — marking quests
|
||||||
|
missed hours early). :meth:`MissedSweepScheduler.settings_changed`
|
||||||
|
wakes the sleeping loop to re-plan whenever the zone or rollover time
|
||||||
|
it planned with has moved, and the loop re-reads the settings before
|
||||||
|
sweeping and re-plans instead if they no longer match its plan.
|
||||||
|
|
||||||
All sweep policy lives in the core (the watermark, the
|
All sweep policy lives in the core (the watermark, the
|
||||||
``[watermark, today)`` window, the no-completion-event rule): this
|
``[watermark, today)`` window, the no-completion-event rule): this
|
||||||
scheduler only decides WHEN. A same-night rerun — e.g. two workers,
|
scheduler only decides WHEN. A same-night rerun — e.g. two workers,
|
||||||
@@ -124,6 +132,11 @@ class MissedSweepScheduler:
|
|||||||
)
|
)
|
||||||
self._sleep = sleep if sleep is not None else asyncio.sleep
|
self._sleep = sleep if sleep is not None else asyncio.sleep
|
||||||
self._task: asyncio.Task | None = None
|
self._task: asyncio.Task | None = None
|
||||||
|
#: ``(timezone, day_rollover_time)`` the pending sweep was
|
||||||
|
#: planned with; ``None`` until the first plan.
|
||||||
|
self._planned: tuple[str, str] | None = None
|
||||||
|
#: Set by :meth:`settings_changed` to cut the current wait short.
|
||||||
|
self._wake = asyncio.Event()
|
||||||
|
|
||||||
def start(self) -> None:
|
def start(self) -> None:
|
||||||
"""Start the background loop (idempotent: a running one stays)."""
|
"""Start the background loop (idempotent: a running one stays)."""
|
||||||
@@ -142,6 +155,18 @@ class MissedSweepScheduler:
|
|||||||
pass
|
pass
|
||||||
self._task = None
|
self._task = None
|
||||||
|
|
||||||
|
def settings_changed(self, settings: object) -> None:
|
||||||
|
"""Re-plan the pending sweep if ``settings`` moved its rollover.
|
||||||
|
|
||||||
|
Called with freshly stored settings (e.g. a zone adopted from
|
||||||
|
Home Assistant). A differing ``timezone`` or
|
||||||
|
``day_rollover_time`` wakes the sleeping loop, which reloads the
|
||||||
|
settings and schedules the next sweep for the new household
|
||||||
|
rollover; an unchanged plan is left sleeping.
|
||||||
|
"""
|
||||||
|
if (settings.timezone, settings.day_rollover_time) != self._planned:
|
||||||
|
self._wake.set()
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def running(self) -> bool:
|
def running(self) -> bool:
|
||||||
"""Whether the background loop task is alive."""
|
"""Whether the background loop task is alive."""
|
||||||
@@ -150,6 +175,9 @@ class MissedSweepScheduler:
|
|||||||
async def _run_forever(self) -> None:
|
async def _run_forever(self) -> None:
|
||||||
"""Schedule → sweep → repeat, forever (until cancelled)."""
|
"""Schedule → sweep → repeat, forever (until cancelled)."""
|
||||||
while True:
|
while True:
|
||||||
|
# Cleared BEFORE the settings read, so a change stored after
|
||||||
|
# the read still wakes this cycle's wait.
|
||||||
|
self._wake.clear()
|
||||||
try:
|
try:
|
||||||
settings = await core_settings_store.load_settings(
|
settings = await core_settings_store.load_settings(
|
||||||
self._database
|
self._database
|
||||||
@@ -168,6 +196,8 @@ class MissedSweepScheduler:
|
|||||||
)
|
)
|
||||||
await self._sleep(_RETRY_BACKOFF_SECONDS)
|
await self._sleep(_RETRY_BACKOFF_SECONDS)
|
||||||
continue
|
continue
|
||||||
|
plan = (settings.timezone, settings.day_rollover_time)
|
||||||
|
self._planned = plan
|
||||||
# The host may run UTC: the rollover is a household wall-clock
|
# The host may run UTC: the rollover is a household wall-clock
|
||||||
# time, so the clock read is re-expressed in the stored zone.
|
# time, so the clock read is re-expressed in the stored zone.
|
||||||
delay = _delay_seconds(
|
delay = _delay_seconds(
|
||||||
@@ -178,8 +208,17 @@ class MissedSweepScheduler:
|
|||||||
settings.day_rollover_time,
|
settings.day_rollover_time,
|
||||||
delay,
|
delay,
|
||||||
)
|
)
|
||||||
await self._sleep(delay)
|
if not await self._sleep_unless_woken(delay):
|
||||||
|
continue
|
||||||
try:
|
try:
|
||||||
|
# Re-read before sweeping: a zone or rollover change the
|
||||||
|
# wake-up missed means this is not the household's
|
||||||
|
# rollover any more, so re-plan instead of sweeping.
|
||||||
|
settings = await core_settings_store.load_settings(
|
||||||
|
self._database
|
||||||
|
)
|
||||||
|
if (settings.timezone, settings.day_rollover_time) != plan:
|
||||||
|
continue
|
||||||
await self._run_once(
|
await self._run_once(
|
||||||
settings.household_now(self._clock()).date()
|
settings.household_now(self._clock()).date()
|
||||||
)
|
)
|
||||||
@@ -194,6 +233,19 @@ class MissedSweepScheduler:
|
|||||||
)
|
)
|
||||||
await self._sleep(_RETRY_BACKOFF_SECONDS)
|
await self._sleep(_RETRY_BACKOFF_SECONDS)
|
||||||
|
|
||||||
|
async def _sleep_unless_woken(self, delay: float) -> bool:
|
||||||
|
"""Wait ``delay``; ``False`` if :meth:`settings_changed` woke it."""
|
||||||
|
sleeper = asyncio.ensure_future(self._sleep(delay))
|
||||||
|
waker = asyncio.ensure_future(self._wake.wait())
|
||||||
|
try:
|
||||||
|
await asyncio.wait(
|
||||||
|
{sleeper, waker}, return_when=asyncio.FIRST_COMPLETED
|
||||||
|
)
|
||||||
|
return sleeper.done()
|
||||||
|
finally:
|
||||||
|
sleeper.cancel()
|
||||||
|
waker.cancel()
|
||||||
|
|
||||||
async def _run_once(self, today: datetime.date) -> None:
|
async def _run_once(self, today: datetime.date) -> None:
|
||||||
"""Run ONE sweep for ``today`` and publish what it built.
|
"""Run ONE sweep for ``today`` and publish what it built.
|
||||||
|
|
||||||
|
|||||||
@@ -31,6 +31,7 @@ from zoneinfo import ZoneInfo
|
|||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from api import routes_admin, routes_panel
|
from api import routes_admin, routes_panel
|
||||||
|
from api import scheduler as scheduler_module
|
||||||
from api.database import _executor
|
from api.database import _executor
|
||||||
from api.scheduler import MissedSweepScheduler, _delay_seconds
|
from api.scheduler import MissedSweepScheduler, _delay_seconds
|
||||||
from tests.admin_jwt_harness import (
|
from tests.admin_jwt_harness import (
|
||||||
@@ -417,17 +418,28 @@ async def test_reported_zone_follows_home_assistant_changes(
|
|||||||
assert (await _stored_settings(household))["timezone"] == HOUSEHOLD_ZONE
|
assert (await _stored_settings(household))["timezone"] == HOUSEHOLD_ZONE
|
||||||
|
|
||||||
|
|
||||||
|
async def _adopt_new_york(household: SimpleNamespace) -> None:
|
||||||
|
"""Prove automatic adoption first, so later assertions can't pass
|
||||||
|
on code that ignores the reported zone altogether."""
|
||||||
|
await _panel_pack_bag(household, HOUSEHOLD_ZONE)
|
||||||
|
stored = await _stored_settings(household)
|
||||||
|
assert stored["timezone"] == HOUSEHOLD_ZONE
|
||||||
|
assert stored["timezone_configured"] is False
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("chosen", ["America/Los_Angeles", ""])
|
@pytest.mark.parametrize("chosen", ["America/Los_Angeles", ""])
|
||||||
async def test_admin_chosen_zone_wins_over_reported_zone(
|
async def test_admin_chosen_zone_wins_over_reported_zone(
|
||||||
household: SimpleNamespace, chosen: str
|
household: SimpleNamespace, chosen: str
|
||||||
) -> None:
|
) -> None:
|
||||||
|
await _adopt_new_york(household)
|
||||||
response = await household.client.patch(
|
response = await household.client.patch(
|
||||||
"/api/v1/admin/settings",
|
"/api/v1/admin/settings",
|
||||||
json={"timezone": chosen, "timezone_configured": True},
|
json={"timezone": chosen, "timezone_configured": True},
|
||||||
headers=_admin_headers(),
|
headers=_admin_headers(),
|
||||||
)
|
)
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
await _panel_pack_bag(household, HOUSEHOLD_ZONE)
|
# A conflicting report is no longer adopted.
|
||||||
|
await _panel_pack_bag(household, "America/Chicago")
|
||||||
stored = await _stored_settings(household)
|
stored = await _stored_settings(household)
|
||||||
assert stored["timezone"] == chosen
|
assert stored["timezone"] == chosen
|
||||||
assert stored["timezone_configured"] is True
|
assert stored["timezone_configured"] is True
|
||||||
@@ -436,8 +448,113 @@ async def test_admin_chosen_zone_wins_over_reported_zone(
|
|||||||
async def test_invalid_reported_zone_is_ignored(
|
async def test_invalid_reported_zone_is_ignored(
|
||||||
household: SimpleNamespace,
|
household: SimpleNamespace,
|
||||||
) -> None:
|
) -> None:
|
||||||
|
await _adopt_new_york(household)
|
||||||
household.clock["now"] = _utc(13, 59, 59)
|
household.clock["now"] = _utc(13, 59, 59)
|
||||||
_today_iso, instance = await _panel_pack_bag(household, "Mars/Olympus_Mons")
|
_today_iso, instance = await _panel_pack_bag(household, "Mars/Olympus_Mons")
|
||||||
# Host-local (UTC) behaviour, unchanged store.
|
# Still read in the adopted zone: 09:59:59 EDT is not overdue.
|
||||||
assert instance["overdue"] is True
|
assert instance["overdue"] is False
|
||||||
assert (await _stored_settings(household))["timezone"] == ""
|
assert (await _stored_settings(household))["timezone"] == HOUSEHOLD_ZONE
|
||||||
|
|
||||||
|
|
||||||
|
# --- the scheduler follows an adopted zone (task 6351b36d) -------------------
|
||||||
|
#
|
||||||
|
# On a UTC host with no stored zone the scheduler first plans for UTC
|
||||||
|
# midnight; a zone adopted while it sleeps must move the sweep to the
|
||||||
|
# household's midnight rather than sweep at 20:00 EDT.
|
||||||
|
|
||||||
|
#: 13:00 UTC (09:00 EDT) to UTC midnight.
|
||||||
|
_UTC_MIDNIGHT_DELAY = 11 * 3600
|
||||||
|
#: 13:00 UTC (09:00 EDT) to New York midnight (04:00 UTC).
|
||||||
|
_NEW_YORK_MIDNIGHT_DELAY = 15 * 3600
|
||||||
|
|
||||||
|
|
||||||
|
async def test_scheduler_replans_when_panel_adopts_zone_while_sleeping(
|
||||||
|
temp_db_path: str,
|
||||||
|
) -> None:
|
||||||
|
runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
|
||||||
|
async with runner as client:
|
||||||
|
app_state = runner.app.state
|
||||||
|
await app_state.sweep_scheduler.stop()
|
||||||
|
delays: list[float] = []
|
||||||
|
slept = asyncio.Condition()
|
||||||
|
|
||||||
|
async def sleep_stub(seconds: float) -> None:
|
||||||
|
async with slept:
|
||||||
|
delays.append(seconds)
|
||||||
|
slept.notify_all()
|
||||||
|
await asyncio.Event().wait()
|
||||||
|
|
||||||
|
scheduler = MissedSweepScheduler(
|
||||||
|
app_state.db.database, clock=lambda: _utc(13, 0), sleep=sleep_stub
|
||||||
|
)
|
||||||
|
app_state.sweep_scheduler = scheduler
|
||||||
|
scheduler.start()
|
||||||
|
try:
|
||||||
|
async with slept:
|
||||||
|
await asyncio.wait_for(
|
||||||
|
slept.wait_for(lambda: len(delays) == 1), timeout=5
|
||||||
|
)
|
||||||
|
response = await client.get(
|
||||||
|
"/api/v1/panel/snapshot",
|
||||||
|
headers=_reporting_panel_headers(HOUSEHOLD_ZONE),
|
||||||
|
)
|
||||||
|
assert response.status_code == 200
|
||||||
|
async with slept:
|
||||||
|
await asyncio.wait_for(
|
||||||
|
slept.wait_for(lambda: len(delays) == 2), timeout=5
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
await scheduler.stop()
|
||||||
|
assert delays == [
|
||||||
|
pytest.approx(_UTC_MIDNIGHT_DELAY),
|
||||||
|
pytest.approx(_NEW_YORK_MIDNIGHT_DELAY),
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
async def test_scheduler_rechecks_settings_before_sweeping(
|
||||||
|
temp_db_path: str, monkeypatch
|
||||||
|
) -> None:
|
||||||
|
"""A zone stored without a wake-up still stops the stale sweep."""
|
||||||
|
database = _core_db.NestQuestDatabase(_executor)
|
||||||
|
await database.open(temp_db_path)
|
||||||
|
try:
|
||||||
|
await _core_migrations.apply_migrations(database)
|
||||||
|
sweeps: list[datetime.date] = []
|
||||||
|
|
||||||
|
async def sweep_stub(_database, *, today):
|
||||||
|
sweeps.append(today)
|
||||||
|
return []
|
||||||
|
|
||||||
|
monkeypatch.setattr(
|
||||||
|
scheduler_module.core_sweep, "run_missed_sweep", sweep_stub
|
||||||
|
)
|
||||||
|
delays: list[float] = []
|
||||||
|
replanned = asyncio.Event()
|
||||||
|
|
||||||
|
async def sleep_stub(seconds: float) -> None:
|
||||||
|
delays.append(seconds)
|
||||||
|
if len(delays) == 1:
|
||||||
|
# The UTC rollover "arrives" after the zone was stored
|
||||||
|
# behind the scheduler's back.
|
||||||
|
await _settings_store.adopt_reported_timezone(
|
||||||
|
database, HOUSEHOLD_ZONE
|
||||||
|
)
|
||||||
|
return
|
||||||
|
replanned.set()
|
||||||
|
await asyncio.Event().wait()
|
||||||
|
|
||||||
|
scheduler = MissedSweepScheduler(
|
||||||
|
database, clock=lambda: _utc(13, 0), sleep=sleep_stub
|
||||||
|
)
|
||||||
|
scheduler.start()
|
||||||
|
try:
|
||||||
|
await asyncio.wait_for(replanned.wait(), timeout=5)
|
||||||
|
finally:
|
||||||
|
await scheduler.stop()
|
||||||
|
assert sweeps == []
|
||||||
|
assert delays == [
|
||||||
|
pytest.approx(_UTC_MIDNIGHT_DELAY),
|
||||||
|
pytest.approx(_NEW_YORK_MIDNIGHT_DELAY),
|
||||||
|
]
|
||||||
|
finally:
|
||||||
|
await database.close()
|
||||||
|
|||||||
@@ -351,6 +351,24 @@ async def test_stream_events_yields_documented_frames_and_sends_bearer() -> None
|
|||||||
assert call["json"] is None
|
assert call["json"] is None
|
||||||
|
|
||||||
|
|
||||||
|
async def test_stream_events_reports_household_timezone() -> None:
|
||||||
|
"""The SSE GET carries Home Assistant's zone beside the Bearer token
|
||||||
|
(task 6351b36d), like the snapshot and completion requests."""
|
||||||
|
frame = _frame_text(*DOCUMENTED_FRAMES[0])
|
||||||
|
transport = _StreamTransport(_StreamingResponse(200, [frame]))
|
||||||
|
client = NestQuestApiClient(
|
||||||
|
BASE_URL, TOKEN, transport, household_timezone="America/New_York"
|
||||||
|
)
|
||||||
|
|
||||||
|
frames = [frame async for frame in client.stream_events()]
|
||||||
|
|
||||||
|
assert frames == [DOCUMENTED_FRAMES[0]]
|
||||||
|
assert transport.calls[0]["headers"] == {
|
||||||
|
"Authorization": f"Bearer {TOKEN}",
|
||||||
|
"X-NestQuest-Timezone": "America/New_York",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
async def test_stream_events_buffers_lines_across_chunks_and_joins_data() -> None:
|
async def test_stream_events_buffers_lines_across_chunks_and_joins_data() -> None:
|
||||||
"""Chunk-split lines parse, and multi-line data joins per SSE rules."""
|
"""Chunk-split lines parse, and multi-line data joins per SSE rules."""
|
||||||
first_type, first_payload = DOCUMENTED_FRAMES[0]
|
first_type, first_payload = DOCUMENTED_FRAMES[0]
|
||||||
|
|||||||
Reference in New Issue
Block a user