diff --git a/nanobot/cron/service.py b/nanobot/cron/service.py index 1cc858ce..c3ccd08a 100644 --- a/nanobot/cron/service.py +++ b/nanobot/cron/service.py @@ -2,8 +2,10 @@ import asyncio import json +import os import time import uuid +from contextlib import suppress from dataclasses import asdict from datetime import datetime from pathlib import Path @@ -83,8 +85,20 @@ class CronService: self._timer_active = False self.max_sleep_ms = max_sleep_ms - def _load_jobs(self) -> tuple[list[CronJob], int]: - jobs = [] + def _load_jobs(self) -> tuple[list[CronJob], int] | None: + """Load jobs from disk. + + Returns: + ``(jobs, version)`` tuple on success or when no store file exists + (in which case an empty list and version 1 are returned). + ``None`` when the store file exists but cannot be parsed; the + corrupt file is preserved with a ``.corrupt-`` suffix so the + caller can decide whether to overwrite or bail out. Returning a + sentinel here is important: silently treating a parse error as an + empty job list would cause the next ``_save_store`` to wipe every + job from disk. + """ + jobs: list[CronJob] = [] version = 1 if self.store_path.exists(): try: @@ -136,7 +150,22 @@ class CronService: delete_after_run=j.get("deleteAfterRun", False), )) except Exception as e: - logger.warning("Failed to load cron store: {}", e) + # Preserve the corrupt file for forensic recovery instead of + # letting the next save overwrite it with an empty job list. + backup = self.store_path.with_suffix( + self.store_path.suffix + f".corrupt-{int(time.time())}" + ) + with suppress(OSError): + self.store_path.rename(backup) + logger.error( + "Failed to load cron store at {}: {}. " + "Corrupt file preserved at {}. " + "Refusing to overwrite to avoid data loss.", + self.store_path, + e, + backup, + ) + return None return jobs, version def _merge_action(self): @@ -175,15 +204,28 @@ class CronService: self._save_store() return - def _load_store(self) -> CronStore: + def _load_store(self) -> CronStore | None: """Load jobs from disk. Reloads automatically if file was modified externally. - Reload every time because it needs to merge operations on the jobs object from other instances. - During _on_timer execution, return the existing store to prevent concurrent _load_store calls (e.g. from list_jobs polling) from replacing it mid-execution. + - When the on-disk store exists but is unreadable: keep using the + previous in-memory ``self._store`` if we already have one (so a + transient corruption does not drop live jobs); only the very first + load (during ``start``) can return ``None`` to signal an unrecoverable + state to the caller. """ if self._timer_active and self._store: return self._store - jobs, version = self._load_jobs() + loaded = self._load_jobs() + if loaded is None: + # Corrupt store on disk. Prefer the last good in-memory snapshot + # over wiping live jobs; ``_load_jobs`` has already moved the + # corrupt file aside with a ``.corrupt-`` suffix. + if self._store is not None: + return self._store + return None + jobs, version = loaded self._store = CronStore(version=version, jobs=jobs) self._merge_action() @@ -242,12 +284,56 @@ class CronService: ] } - self.store_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") + self._atomic_write(self.store_path, json.dumps(data, indent=2, ensure_ascii=False)) + + @staticmethod + def _atomic_write(path: Path, content: str) -> None: + """Write *content* to *path* atomically with fsync. + + Uses a temp-file + ``os.replace`` + ``fsync`` pattern so a crash or + SIGKILL mid-write cannot leave the destination truncated or invalid. + Mirrors ``nanobot.session.manager.SessionManager.save`` (see + commit 512bf59, ``fix(session): fsync sessions on graceful shutdown + to prevent data loss``). Without this, ``jobs.json`` could be + corrupted on container shutdown and silently re-created empty on + next start, wiping every scheduled job. + """ + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = path.with_suffix(path.suffix + ".tmp") + try: + with open(tmp_path, "w", encoding="utf-8") as f: + f.write(content) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp_path, path) + # fsync the parent directory so the rename itself is durable. + # Skip on Windows where opening a directory raises PermissionError; + # NTFS journals metadata synchronously so this is a no-op there. + with suppress(PermissionError): + fd = os.open(str(path.parent), os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + except BaseException: + tmp_path.unlink(missing_ok=True) + raise async def start(self) -> None: """Start the cron service.""" self._running = True - self._load_store() + loaded = self._load_store() + if loaded is None: + # Store file existed but was corrupt and has been preserved with + # a ``.corrupt-`` suffix. Bail out instead of starting with + # an empty store; that would call ``_save_store`` and overwrite + # the now-renamed (but still recoverable) data with []. + self._running = False + raise RuntimeError( + f"cron store at {self.store_path} is corrupt and was preserved; " + "refusing to start with an empty job list. " + "Inspect the .corrupt- backup and restore manually." + ) self._recompute_next_runs() self._save_store() self._arm_timer() @@ -301,7 +387,11 @@ class CronService: async def _on_timer(self) -> None: """Handle timer tick - run due jobs.""" - self._load_store() + loaded = self._load_store() + # If a hot reload found a corrupt store on disk, ``loaded`` is + # ``None`` but ``self._store`` may still hold the previous, + # known-good in-memory snapshot. Keep using it rather than + # crashing the timer or wiping live jobs. if not self._store: self._arm_timer() return diff --git a/tests/cron/test_cron_persistence.py b/tests/cron/test_cron_persistence.py new file mode 100644 index 00000000..fb218687 --- /dev/null +++ b/tests/cron/test_cron_persistence.py @@ -0,0 +1,167 @@ +"""Persistence tests for ``nanobot.cron.service.CronService``. + +These tests target the specific failure mode where a corrupt or partially +written ``jobs.json`` would silently turn into an empty job list on the next +start, deleting every scheduled job. See ``fix(cron): atomic write for +jobs.json + don't silently overwrite corrupt store``. +""" + +from __future__ import annotations + +import json +import time +from pathlib import Path + +import pytest + +from nanobot.cron.service import CronService +from nanobot.cron.types import CronSchedule + + +def _seeded_store(tmp_path: Path) -> tuple[CronService, Path]: + """Build a service with one persisted job on disk and return both the + service and the resolved store path. Adds the job via the action log + (the path used when the service is not running) and then triggers a + merge so ``jobs.json`` is written, mirroring the persisted on-disk + state seen in production.""" + store_path = tmp_path / "cron" / "jobs.json" + service = CronService(store_path) + service.add_job( + name="Daily Loving Message", + schedule=CronSchedule(kind="cron", expr="0 10 * * *", tz="Asia/Kuwait"), + message="hello", + ) + # add_job appended to action.jsonl; flush to jobs.json by toggling + # ``_running`` long enough for ``_merge_action`` to do its rewrite. + service._running = True + try: + service._load_store() + finally: + service._running = False + assert store_path.exists() + return service, store_path + + +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.""" + service, store_path = _seeded_store(tmp_path) + + # Simulate an arbitrary save and confirm the result parses cleanly and + # no orphan ``.tmp`` is left behind. + service._save_store() + data = json.loads(store_path.read_text(encoding="utf-8")) + assert len(data["jobs"]) == 1 + + tmp_files = list(store_path.parent.glob("*.tmp")) + assert tmp_files == [], f"unexpected temp files left behind: {tmp_files}" + + +def test_save_store_failure_does_not_corrupt_existing_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """If writing the temp file blows up partway through, the previous + ``jobs.json`` must remain readable. This is the regression we are + actually fixing: pre-fix, ``write_text`` would truncate the destination + in place and leave it corrupt.""" + service, store_path = _seeded_store(tmp_path) + original = store_path.read_bytes() + + # Inject a failure inside the temp-file write. ``os.replace`` should + # never run; the destination must keep its previous content. + real_open = open + + def boom(path, *args, **kwargs): # type: ignore[no-untyped-def] + if str(path).endswith(".tmp"): + raise OSError("simulated disk full") + return real_open(path, *args, **kwargs) + + monkeypatch.setattr("builtins.open", boom) + + with pytest.raises(OSError, match="simulated disk full"): + service._save_store() + + assert store_path.read_bytes() == original + + +def test_load_jobs_preserves_corrupt_store_and_returns_none( + tmp_path: Path, +) -> None: + """A corrupt ``jobs.json`` must not be silently treated as an empty + list. The loader returns ``None`` and the corrupt file is moved aside + with a ``.corrupt-`` suffix so an operator can recover it.""" + store_path = tmp_path / "cron" / "jobs.json" + store_path.parent.mkdir(parents=True) + store_path.write_text("{not valid json", encoding="utf-8") + + service = CronService(store_path) + assert service._load_jobs() is None + + # Original path is gone; a ``.corrupt-`` backup exists alongside it. + 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 test_start_refuses_to_overwrite_corrupt_store(tmp_path: Path) -> None: + """``start`` must abort instead of running ``_save_store`` against an + empty in-memory state when the on-disk store is corrupt. Otherwise the + next save would overwrite the (recoverable) corrupt file with an empty + job list and the user's jobs would be unrecoverable.""" + store_path = tmp_path / "cron" / "jobs.json" + store_path.parent.mkdir(parents=True) + store_path.write_text("{still not json", encoding="utf-8") + + service = CronService(store_path) + import asyncio + + with pytest.raises(RuntimeError, match="corrupt"): + asyncio.run(service.start()) + + # Service is left in a stopped state so the operator notices. + assert service._running is False + + # And the corrupt file is still recoverable from the .corrupt- copy. + backups = list(store_path.parent.glob("jobs.json.corrupt-*")) + assert len(backups) == 1 + + +def test_load_store_falls_back_to_in_memory_on_corruption_after_start( + tmp_path: Path, +) -> None: + """If the store file becomes corrupt *after* a successful start (e.g. a + rclone-mounted Drive returns a partial read), the service must keep + using its existing in-memory snapshot instead of dropping every job.""" + service, store_path = _seeded_store(tmp_path) + # Force load so ``self._store`` is populated. + service._load_store() + snapshot = service._store + assert snapshot is not None and len(snapshot.jobs) == 1 + + # Now corrupt the file on disk. + store_path.write_text("\x00garbage\x00", encoding="utf-8") + + # Subsequent reload returns the in-memory snapshot, not None or empty. + result = service._load_store() + assert result is snapshot + assert len(result.jobs) == 1 + assert result.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.""" + store_path = tmp_path / "cron" / "jobs.json" + + s1 = CronService(store_path) + s1.add_job( + name="Daily Loving Message", + schedule=CronSchedule(kind="cron", expr="0 10 * * *", tz="Asia/Kuwait"), + message="hello", + ) + + s2 = CronService(store_path) + s2._load_store() + assert s2._store is not None + assert [j.name for j in s2._store.jobs] == ["Daily Loving Message"]