From e5dbb15c344ad9225b6e980d34643b572716b3c0 Mon Sep 17 00:00:00 2001 From: yorkhellen Date: Sun, 28 Jun 2026 00:10:12 +0800 Subject: [PATCH] fix(cron): guard public APIs against unavailable store --- nanobot/cron/service.py | 37 +++++-- tests/cron/test_cron_persistence.py | 146 +++++++++++++++++++++++++++- 2 files changed, 173 insertions(+), 10 deletions(-) diff --git a/nanobot/cron/service.py b/nanobot/cron/service.py index 23426bec..b42373cc 100644 --- a/nanobot/cron/service.py +++ b/nanobot/cron/service.py @@ -357,6 +357,25 @@ class CronService: return self._store + def _require_store(self) -> CronStore: + """Return a usable store or raise a clear error. + + ``_load_store`` deliberately returns ``None`` when the first load sees + a corrupt on-disk store and no previous in-memory snapshot exists. The + public API requires a concrete store object before touching + ``store.jobs``; raising here keeps callers from seeing an accidental + ``AttributeError`` and, more importantly, prevents follow-up saves from + treating a corrupt store as an empty one. + """ + store = self._load_store() + if store is None: + raise RuntimeError( + f"cron store at {self.store_path} could not be loaded and was preserved " + "as a .corrupt- backup; refusing to operate to avoid overwriting " + "scheduled jobs. Inspect the corrupt backup and restore jobs.json manually." + ) + return store + def _save_store(self) -> None: """Save jobs to disk.""" if not self._store: @@ -622,7 +641,7 @@ class CronService: def list_jobs(self, include_disabled: bool = False) -> list[CronJob]: """List all jobs.""" - store = self._load_store() + store = self._require_store() jobs = store.jobs if include_disabled else [j for j in store.jobs if j.enabled] return sorted(jobs, key=lambda j: j.state.next_run_at_ms or float('inf')) @@ -684,7 +703,7 @@ class CronService: _normalize_agent_turn_job(job) self._enforce_agent_binding(job) if self._running: - store = self._load_store() + store = self._require_store() store.jobs.append(job) self._save_store() self._arm_timer() @@ -696,7 +715,7 @@ class CronService: def register_system_job(self, job: CronJob) -> CronJob: """Register an internal system job (idempotent on restart).""" - store = self._load_store() + store = self._require_store() now = _now_ms() job.state = CronJobState(next_run_at_ms=_compute_next_run(job.schedule, now)) job.created_at_ms = now @@ -710,7 +729,7 @@ class CronService: def remove_job(self, job_id: str) -> Literal["removed", "protected", "not_found"]: """Remove a job by ID, unless it is a protected system job.""" - store = self._load_store() + store = self._require_store() job = next((j for j in store.jobs if j.id == job_id), None) if job is None: return "not_found" @@ -735,7 +754,7 @@ class CronService: def enable_job(self, job_id: str, enabled: bool = True) -> CronJob | None: """Enable or disable a job.""" - store = self._load_store() + store = self._require_store() for job in store.jobs: if job.id == job_id: job.enabled = enabled @@ -770,7 +789,7 @@ class CronService: For ``channel`` and ``to``, pass an explicit value (including ``None``) to update; omit (sentinel ``...``) to leave unchanged. """ - store = self._load_store() + store = self._require_store() job = next((j for j in store.jobs if j.id == job_id), None) if job is None: return "not_found" @@ -815,7 +834,7 @@ class CronService: was_running = self._running self._running = True try: - store = self._load_store() + store = self._require_store() for job in store.jobs: if job.id == job_id: if self._is_unbound_agent_job(job): @@ -835,12 +854,12 @@ class CronService: def get_job(self, job_id: str) -> CronJob | None: """Get a job by ID.""" - store = self._load_store() + store = self._require_store() return next((j for j in store.jobs if j.id == job_id), None) def status(self) -> dict: """Get service status.""" - store = self._load_store() + store = self._require_store() return { "enabled": self._running, "jobs": len(store.jobs), diff --git a/tests/cron/test_cron_persistence.py b/tests/cron/test_cron_persistence.py index 4732f61e..863ff0d9 100644 --- a/tests/cron/test_cron_persistence.py +++ b/tests/cron/test_cron_persistence.py @@ -10,11 +10,12 @@ from __future__ import annotations import json from pathlib import Path +from typing import Callable import pytest from nanobot.cron.service import CronService -from nanobot.cron.types import CronSchedule +from nanobot.cron.types import CronJob, CronPayload, CronSchedule def _seeded_store(tmp_path: Path) -> tuple[CronService, Path]: @@ -41,6 +42,29 @@ def _seeded_store(tmp_path: Path) -> tuple[CronService, Path]: return service, store_path +def _corrupt_store(tmp_path: Path) -> Path: + store_path = tmp_path / "cron" / "jobs.json" + store_path.parent.mkdir(parents=True) + store_path.write_text("{not valid json", encoding="utf-8") + return store_path + + +def _assert_single_corrupt_backup(store_path: Path) -> None: + assert not store_path.exists() + backups = list(store_path.parent.glob("jobs.json.corrupt-*")) + assert len(backups) == 1 + assert backups[0].read_text(encoding="utf-8") == "{not valid json" + + +def _system_job(job_id: str = "dream") -> CronJob: + return CronJob( + id=job_id, + name="Dream", + schedule=CronSchedule(kind="cron", expr="0 */2 * * *", tz="UTC"), + payload=CronPayload(kind="system_event"), + ) + + def test_save_store_is_atomic(tmp_path: Path) -> None: """``_save_store`` must use temp-file + rename so an interrupted write cannot leave the destination truncated or invalid.""" @@ -148,6 +172,126 @@ def test_load_store_falls_back_to_in_memory_on_corruption_after_start( assert result.jobs[0].name == "Daily Loving Message" +@pytest.mark.parametrize( + ("api_name", "call"), + [ + ("list_jobs", lambda service: service.list_jobs()), + ("get_job", lambda service: service.get_job("missing")), + ("status", lambda service: service.status()), + ("remove_job", lambda service: service.remove_job("missing")), + ("enable_job", lambda service: service.enable_job("missing", enabled=False)), + ("update_job", lambda service: service.update_job("missing", name="new name")), + ("register_system_job", lambda service: service.register_system_job(_system_job())), + ], +) +def test_public_apis_raise_clear_error_for_unavailable_corrupt_store( + tmp_path: Path, + api_name: str, + call: Callable[[CronService], object], +) -> None: + """Public APIs should report the corrupt store explicitly instead of + leaking ``AttributeError`` when the first load cannot produce a store.""" + store_path = _corrupt_store(tmp_path) + service = CronService(store_path) + + with pytest.raises(RuntimeError, match="corrupt.*restore jobs.json") as exc_info: + call(service) + + assert api_name + assert str(store_path) in str(exc_info.value) + _assert_single_corrupt_backup(store_path) + + +@pytest.mark.asyncio +async def test_run_job_raises_clear_error_and_restores_running_state_for_corrupt_store( + tmp_path: Path, +) -> None: + store_path = _corrupt_store(tmp_path) + service = CronService(store_path) + + with pytest.raises(RuntimeError, match="corrupt.*restore jobs.json"): + await service.run_job("missing") + + assert service._running is False + _assert_single_corrupt_backup(store_path) + + +@pytest.mark.asyncio +async def test_run_job_preserves_running_state_when_corrupt_store_unavailable( + tmp_path: Path, +) -> None: + store_path = _corrupt_store(tmp_path) + service = CronService(store_path) + service._running = True + service._arm_timer = lambda: None + + with pytest.raises(RuntimeError, match="corrupt.*restore jobs.json"): + await service.run_job("missing") + + assert service._running is True + service.stop() + + +def test_running_add_job_raises_clear_error_for_unavailable_corrupt_store( + tmp_path: Path, +) -> None: + store_path = _corrupt_store(tmp_path) + service = CronService(store_path) + service._running = True + + with pytest.raises(RuntimeError, match="corrupt.*restore jobs.json"): + service.add_job( + name="running add", + schedule=CronSchedule(kind="every", every_ms=60_000), + message="hello", + session_key="websocket:chat-1", + origin_channel="websocket", + origin_chat_id="chat-1", + ) + + _assert_single_corrupt_backup(store_path) + + +def test_stopped_add_job_still_appends_action_without_loading_corrupt_store( + tmp_path: Path, +) -> None: + """The stopped-service add path is an action-log write and must not start + requiring a readable store.""" + store_path = _corrupt_store(tmp_path) + service = CronService(store_path) + + job = service.add_job( + name="offline add", + schedule=CronSchedule(kind="every", every_ms=60_000), + message="hello", + session_key="websocket:chat-1", + origin_channel="websocket", + origin_chat_id="chat-1", + ) + + assert job.name == "offline add" + assert store_path.exists() + assert store_path.read_text(encoding="utf-8") == "{not valid json" + assert list(store_path.parent.glob("jobs.json.corrupt-*")) == [] + actions = (store_path.parent / "action.jsonl").read_text(encoding="utf-8").splitlines() + assert len(actions) == 1 + assert json.loads(actions[0])["action"] == "add" + + +def test_public_api_uses_in_memory_snapshot_when_disk_becomes_corrupt( + tmp_path: Path, +) -> None: + service, store_path = _seeded_store(tmp_path) + service._load_store() + assert service._store is not None + store_path.write_text("{not valid json", encoding="utf-8") + + jobs = service.list_jobs(include_disabled=True) + + assert len(jobs) == 1 + assert jobs[0].name == "Daily Loving Message" + + def test_full_round_trip_survives_repeated_save_load(tmp_path: Path) -> None: """Sanity check: jobs survive add → save → reload across fresh ``CronService`` instances pointing at the same store."""