fix(cron): guard public APIs against unavailable store

This commit is contained in:
yorkhellen
2026-06-28 19:46:44 +08:00
committed by Xubin Ren
parent c90e433057
commit e5dbb15c34
2 changed files with 173 additions and 10 deletions
+28 -9
View File
@@ -357,6 +357,25 @@ class CronService:
return self._store 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-<ts> 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: def _save_store(self) -> None:
"""Save jobs to disk.""" """Save jobs to disk."""
if not self._store: if not self._store:
@@ -622,7 +641,7 @@ class CronService:
def list_jobs(self, include_disabled: bool = False) -> list[CronJob]: def list_jobs(self, include_disabled: bool = False) -> list[CronJob]:
"""List all jobs.""" """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] 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')) 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) _normalize_agent_turn_job(job)
self._enforce_agent_binding(job) self._enforce_agent_binding(job)
if self._running: if self._running:
store = self._load_store() store = self._require_store()
store.jobs.append(job) store.jobs.append(job)
self._save_store() self._save_store()
self._arm_timer() self._arm_timer()
@@ -696,7 +715,7 @@ class CronService:
def register_system_job(self, job: CronJob) -> CronJob: def register_system_job(self, job: CronJob) -> CronJob:
"""Register an internal system job (idempotent on restart).""" """Register an internal system job (idempotent on restart)."""
store = self._load_store() store = self._require_store()
now = _now_ms() now = _now_ms()
job.state = CronJobState(next_run_at_ms=_compute_next_run(job.schedule, now)) job.state = CronJobState(next_run_at_ms=_compute_next_run(job.schedule, now))
job.created_at_ms = now job.created_at_ms = now
@@ -710,7 +729,7 @@ class CronService:
def remove_job(self, job_id: str) -> Literal["removed", "protected", "not_found"]: def remove_job(self, job_id: str) -> Literal["removed", "protected", "not_found"]:
"""Remove a job by ID, unless it is a protected system job.""" """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) job = next((j for j in store.jobs if j.id == job_id), None)
if job is None: if job is None:
return "not_found" return "not_found"
@@ -735,7 +754,7 @@ class CronService:
def enable_job(self, job_id: str, enabled: bool = True) -> CronJob | None: def enable_job(self, job_id: str, enabled: bool = True) -> CronJob | None:
"""Enable or disable a job.""" """Enable or disable a job."""
store = self._load_store() store = self._require_store()
for job in store.jobs: for job in store.jobs:
if job.id == job_id: if job.id == job_id:
job.enabled = enabled job.enabled = enabled
@@ -770,7 +789,7 @@ class CronService:
For ``channel`` and ``to``, pass an explicit value (including ``None``) For ``channel`` and ``to``, pass an explicit value (including ``None``)
to update; omit (sentinel ``...``) to leave unchanged. 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) job = next((j for j in store.jobs if j.id == job_id), None)
if job is None: if job is None:
return "not_found" return "not_found"
@@ -815,7 +834,7 @@ class CronService:
was_running = self._running was_running = self._running
self._running = True self._running = True
try: try:
store = self._load_store() store = self._require_store()
for job in store.jobs: for job in store.jobs:
if job.id == job_id: if job.id == job_id:
if self._is_unbound_agent_job(job): if self._is_unbound_agent_job(job):
@@ -835,12 +854,12 @@ class CronService:
def get_job(self, job_id: str) -> CronJob | None: def get_job(self, job_id: str) -> CronJob | None:
"""Get a job by ID.""" """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) return next((j for j in store.jobs if j.id == job_id), None)
def status(self) -> dict: def status(self) -> dict:
"""Get service status.""" """Get service status."""
store = self._load_store() store = self._require_store()
return { return {
"enabled": self._running, "enabled": self._running,
"jobs": len(store.jobs), "jobs": len(store.jobs),
+145 -1
View File
@@ -10,11 +10,12 @@ from __future__ import annotations
import json import json
from pathlib import Path from pathlib import Path
from typing import Callable
import pytest import pytest
from nanobot.cron.service import CronService 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]: 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 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: def test_save_store_is_atomic(tmp_path: Path) -> None:
"""``_save_store`` must use temp-file + rename so an interrupted write """``_save_store`` must use temp-file + rename so an interrupted write
cannot leave the destination truncated or invalid.""" 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" 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: def test_full_round_trip_survives_repeated_save_load(tmp_path: Path) -> None:
"""Sanity check: jobs survive add → save → reload across fresh """Sanity check: jobs survive add → save → reload across fresh
``CronService`` instances pointing at the same store.""" ``CronService`` instances pointing at the same store."""