diff --git a/api/routes_panel.py b/api/routes_panel.py index 9e06cc1..16676e1 100644 --- a/api/routes_panel.py +++ b/api/routes_panel.py @@ -200,17 +200,21 @@ async def _household_clock( (:func:`nestquest_core.settings_store.adopt_reported_timezone`), so 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 + it re-plans for the household rollover — under the scheduler's + ``settings_lock``, so the adoption never lands between its settings + check and its sweep. An invalid reported zone is logged and ignored — the panel keeps serving on the stored settings. """ reported = request.headers.get(PANEL_TIMEZONE_HEADER, "").strip() if reported: + scheduler = request.app.state.sweep_scheduler try: - settings = await core_settings_store.adopt_reported_timezone( - database, reported - ) - request.app.state.sweep_scheduler.settings_changed(settings) + async with scheduler.settings_lock: + settings = await core_settings_store.adopt_reported_timezone( + database, reported + ) + scheduler.settings_changed(settings) except ValueError: LOGGER.warning( "Ignoring invalid %s header value %r", diff --git a/api/scheduler.py b/api/scheduler.py index e97de58..e755a7c 100644 --- a/api/scheduler.py +++ b/api/scheduler.py @@ -21,6 +21,19 @@ 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. +Guarantee: no sweep runs for a plan a committed adoption has replaced. +An adopting writer holds :attr:`MissedSweepScheduler.settings_lock` +across its write AND its :meth:`~MissedSweepScheduler.settings_changed` +call; the loop holds the same lock across its plan read (read, record +the plan, clear the wake flag) and across the pre-sweep re-read plus +the sweep itself. So an adoption commits either before a plan read +(which then sees it), after one (``settings_changed`` compares against +that exact plan and wakes the wait), or after the sweep — never between +a read and what the loop does with it. The lock is never held across +the wait, and the loop takes nothing else before it, so a waiting +panel request cannot deadlock: it is delayed at most by one settings +read or one sweep. + 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, @@ -137,6 +150,9 @@ class MissedSweepScheduler: self._planned: tuple[str, str] | None = None #: Set by :meth:`settings_changed` to cut the current wait short. self._wake = asyncio.Event() + #: Serialises settings adoption with planning and sweeping (see + #: the module docstring's guarantee). + self.settings_lock = asyncio.Lock() def start(self) -> None: """Start the background loop (idempotent: a running one stays).""" @@ -159,7 +175,9 @@ class MissedSweepScheduler: """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 + Home Assistant), with :attr:`settings_lock` held across the write + and this call so the comparison is against the loop's current + plan, never a read still in flight. 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. @@ -175,16 +193,19 @@ class MissedSweepScheduler: async def _run_forever(self) -> None: """Schedule → sweep → repeat, forever (until cancelled).""" while True: - # Cleared BEFORE the settings read, so a change stored after - # the read still wakes this cycle's wait. - self._wake.clear() try: - settings = await core_settings_store.load_settings( - self._database - ) - rollover = datetime.time.fromisoformat( - settings.day_rollover_time - ) + # Under the lock no adoption can commit between this read + # and the recorded plan, so any later one wakes the wait. + async with self.settings_lock: + self._wake.clear() + settings = await core_settings_store.load_settings( + self._database + ) + rollover = datetime.time.fromisoformat( + settings.day_rollover_time + ) + plan = (settings.timezone, settings.day_rollover_time) + self._planned = plan except Exception: # Not silent: the failure is logged and retried after # the backoff, so one bad settings document or a @@ -196,8 +217,6 @@ class MissedSweepScheduler: ) await self._sleep(_RETRY_BACKOFF_SECONDS) continue - plan = (settings.timezone, settings.day_rollover_time) - self._planned = plan # The host may run UTC: the rollover is a household wall-clock # time, so the clock read is re-expressed in the stored zone. delay = _delay_seconds( @@ -208,24 +227,30 @@ class MissedSweepScheduler: settings.day_rollover_time, delay, ) - if not await self._sleep_unless_woken(delay): - continue try: + if not await self._sleep_unless_woken(delay): + continue # 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( - settings.household_now(self._clock()).date() - ) + # rollover any more, so re-plan instead of sweeping. The + # lock keeps an adoption from committing between this + # check and the sweep. + async with self.settings_lock: + settings = await core_settings_store.load_settings( + self._database + ) + if ( + settings.timezone, + settings.day_rollover_time, + ) != plan: + continue + await self._run_once( + settings.household_now(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. + # Not silent either: a failed wait or sweep 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", @@ -234,17 +259,27 @@ class MissedSweepScheduler: await self._sleep(_RETRY_BACKOFF_SECONDS) async def _sleep_unless_woken(self, delay: float) -> bool: - """Wait ``delay``; ``False`` if :meth:`settings_changed` woke it.""" + """Wait ``delay``; ``False`` if :meth:`settings_changed` woke it. + + Both children are cancelled AND awaited before returning (or + before a cancellation propagates), so none outlives :meth:`stop`; + a sleep that failed re-raises its exception here for the loop's + retry policy. + """ 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: + slept = sleeper.done() sleeper.cancel() waker.cancel() + await asyncio.gather(sleeper, waker, return_exceptions=True) + if slept: + sleeper.result() + return slept async def _run_once(self, today: datetime.date) -> None: """Run ONE sweep for ``today`` and publish what it built. diff --git a/tests/test_api_household_clock.py b/tests/test_api_household_clock.py index 2879403..7a158f9 100644 --- a/tests/test_api_household_clock.py +++ b/tests/test_api_household_clock.py @@ -558,3 +558,250 @@ async def test_scheduler_rechecks_settings_before_sweeping( ] finally: await database.close() + + +# --- adoption is serialised with planning and sweeping (TZ-003) -------------- +# +# An adoption racing the loop's own settings reads must never leave a +# stale plan behind. Each test holds the loop at one of its reads and +# issues a real panel adoption there; when the scheduler's settings lock +# is held at that point the adoption provably cannot commit until the +# loop moves on, otherwise it is awaited to commit inside the window. + + +def _hold_scheduler_read(monkeypatch, scheduler, nth: int, during): + """Run ``during(settings)`` inside the scheduler task's ``nth`` read.""" + real_load = _settings_store.load_settings + reads = {"count": 0} + + async def load_settings(database): + settings = await real_load(database) + if asyncio.current_task() is scheduler._task: + reads["count"] += 1 + if reads["count"] == nth: + await during(settings) + return settings + + monkeypatch.setattr( + scheduler_module.core_settings_store, "load_settings", load_settings + ) + return real_load + + +async def _adopt_via(client, zone: str) -> None: + response = await client.get( + "/api/v1/panel/snapshot", headers=_reporting_panel_headers(zone) + ) + assert response.status_code == 200 + + +async def _adopt_inside_read(client, scheduler, zone: str) -> asyncio.Task: + """Start a panel adoption of ``zone``; let it commit unless serialised.""" + adoption = asyncio.create_task( + client.get( + "/api/v1/panel/snapshot", headers=_reporting_panel_headers(zone) + ) + ) + lock = getattr(scheduler, "settings_lock", None) + if lock is None or not lock.locked(): + await adoption + return adoption + + +async def test_scheduler_follows_successive_zone_changes( + temp_db_path: str, monkeypatch +) -> None: + """New York -> UTC -> New York while the re-plan read is in flight.""" + 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 + await _adopt_via(client, HOUSEHOLD_ZONE) + adoptions: list[asyncio.Task] = [] + + async def back_to_new_york(_settings) -> None: + adoptions.append( + await _adopt_inside_read(client, scheduler, HOUSEHOLD_ZONE) + ) + + # Read 1 plans New York; read 2 is the re-plan after UTC. + _hold_scheduler_read(monkeypatch, scheduler, 2, back_to_new_york) + scheduler.start() + try: + async with slept: + await asyncio.wait_for( + slept.wait_for(lambda: len(delays) == 1), timeout=5 + ) + await _adopt_via(client, "UTC") + async with slept: + await asyncio.wait_for( + slept.wait_for(lambda: len(delays) == 3), timeout=5 + ) + (adoption,) = adoptions + assert (await adoption).status_code == 200 + finally: + await scheduler.stop() + # The in-flight read planned UTC; the New York adoption that followed + # it woke the wait, so the scheduler sleeps on New York midnight. + assert delays == [ + pytest.approx(_NEW_YORK_MIDNIGHT_DELAY), + pytest.approx(_UTC_MIDNIGHT_DELAY), + pytest.approx(_NEW_YORK_MIDNIGHT_DELAY), + ] + + +async def test_adoption_after_presweep_check_cannot_precede_the_sweep( + temp_db_path: str, monkeypatch +) -> None: + """No sweep runs once an adoption has replaced its plan.""" + 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] = [] + replanned = asyncio.Event() + + async def sleep_stub(seconds: float) -> None: + delays.append(seconds) + if len(delays) == 1: + return # the UTC rollover arrives + replanned.set() + await asyncio.Event().wait() + + scheduler = MissedSweepScheduler( + app_state.db.database, clock=lambda: _utc(13, 0), sleep=sleep_stub + ) + app_state.sweep_scheduler = scheduler + adoptions: list[asyncio.Task] = [] + + async def adopt_new_york(_settings) -> None: + adoptions.append( + await _adopt_inside_read(client, scheduler, HOUSEHOLD_ZONE) + ) + + # Read 1 plans UTC; read 2 is the pre-sweep re-check. + real_load = _hold_scheduler_read( + monkeypatch, scheduler, 2, adopt_new_york + ) + swept_under: list[str] = [] + + async def sweep_stub(database, *, today): + swept_under.append((await real_load(database)).timezone) + return [] + + monkeypatch.setattr( + scheduler_module.core_sweep, "run_missed_sweep", sweep_stub + ) + scheduler.start() + try: + await asyncio.wait_for(replanned.wait(), timeout=5) + (adoption,) = adoptions + assert (await adoption).status_code == 200 + finally: + await scheduler.stop() + # The sweep ran for the UTC plan with UTC still stored, never after + # New York had been adopted; the next plan is New York midnight. + assert swept_under == [""] + assert delays == [ + pytest.approx(_UTC_MIDNIGHT_DELAY), + pytest.approx(_NEW_YORK_MIDNIGHT_DELAY), + ] + + +# --- the wait's child tasks (TZ-004) ------------------------------------------ + + +async def test_failed_sleep_is_retried_not_swept( + temp_db_path: str, monkeypatch +) -> None: + """A sleep that raises backs off and re-plans; nothing is swept.""" + 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: + raise RuntimeError("timer failed") + if len(delays) == 2: + return # the retry backoff + 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), + scheduler_module._RETRY_BACKOFF_SECONDS, + pytest.approx(_UTC_MIDNIGHT_DELAY), + ] + finally: + await database.close() + + +async def test_stop_waits_for_the_sleep_to_finish_cancelling( + temp_db_path: str, +) -> None: + database = _core_db.NestQuestDatabase(_executor) + await database.open(temp_db_path) + try: + await _core_migrations.apply_migrations(database) + sleeping = asyncio.Event() + sleepers: list[asyncio.Task] = [] + cleaned_up: list[bool] = [] + + async def sleep_stub(_seconds: float) -> None: + sleepers.append(asyncio.current_task()) + sleeping.set() + try: + await asyncio.Event().wait() + finally: + # Cancellation cleanup that takes a loop turn. + await asyncio.sleep(0) + cleaned_up.append(True) + + scheduler = MissedSweepScheduler( + database, clock=lambda: _utc(13, 0), sleep=sleep_stub + ) + scheduler.start() + try: + await asyncio.wait_for(sleeping.wait(), timeout=5) + finally: + await scheduler.stop() + (sleeper,) = sleepers + assert sleeper.done() + assert cleaned_up == [True] + finally: + await database.close()