Serialise time zone adoption with the missed-sweep scheduler

The wake flag compared an adoption only against the plan the loop last
recorded, so New York -> UTC -> New York during the re-plan read left
the scheduler sleeping on UTC midnight; and an adoption committing
between the pre-sweep check and the sweep still let the stale sweep run.

`MissedSweepScheduler.settings_lock` now serialises the panel's adoption
(write plus `settings_changed`) with the loop's plan read and with its
pre-sweep re-read plus sweep.  It is never held across the wait, and the
loop holds nothing else while acquiring it, so a panel request is at
most delayed by one settings read or one sweep.

`_sleep_unless_woken` now cancels and awaits both child tasks, and
re-raises a failed sleep inside the loop's retry block, so the backoff
runs instead of a sweep and nothing outlives `stop()`.

Tests (all four fail on bcf387d): the successive-zone race, an adoption
issued during the pre-sweep check, a failing sleep, and stop() waiting
for the sleep's cancellation cleanup.

Task: 6351b36d-0eef-4e7d-bf5a-96b754cbc883

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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()