diff --git a/nanobot/cron/service.py b/nanobot/cron/service.py index b505bea2a..76bd5b851 100644 --- a/nanobot/cron/service.py +++ b/nanobot/cron/service.py @@ -170,6 +170,7 @@ class CronService: self._timer_task: asyncio.Task[None] | None = None self._running = False self._active_executions = 0 + self._store_dirty = False self.max_sleep_ms = max_sleep_ms def _should_persist_store(self) -> bool: @@ -305,6 +306,11 @@ class CronService: load (during ``start``) can return ``None`` to signal an unrecoverable state to the caller. """ + # Never replace state that a previous save failed to persist. Reloading + # the older on-disk snapshot here could make an already executed job due + # again and repeat its side effect. + if self._store_dirty and self._store: + return self._store if self._active_executions > 0 and self._store and not reload_during_execution: return self._store loaded = self._load_jobs() @@ -347,6 +353,9 @@ class CronService: if not self._store: return + # Set this before serialization/write so every exceptional exit keeps + # the in-memory snapshot authoritative until a later save succeeds. + self._store_dirty = True self.store_path.parent.mkdir(parents=True, exist_ok=True) data = { @@ -399,6 +408,7 @@ class CronService: } self._atomic_write(self.store_path, json.dumps(data, indent=2, ensure_ascii=False)) + self._store_dirty = False @staticmethod def _atomic_write(path: Path, content: str) -> None: @@ -514,10 +524,17 @@ class CronService: reload_store = self._active_executions == 0 self._active_executions += 1 try: + # A prior tick may have completed external side effects but failed + # to persist their advanced schedule. Persist that exact snapshot + # before reloading or executing anything else; otherwise the older + # disk state can replay the same job. + if self._store_dirty: + self._save_store() + return + store = self._load_store(reload_during_execution=reload_store) # If a hot reload found a corrupt store on disk, ``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 store is None: return @@ -808,6 +825,11 @@ class CronService: reload_store = self._active_executions == 0 self._active_executions += 1 try: + # A manual run is another side-effecting entrypoint. Do not start + # it while the result of a previous timer execution is still only + # in memory. + if self._store_dirty: + self._save_store() store = self._require_store(reload_during_execution=reload_store) for job in store.jobs: if job.id == job_id: diff --git a/tests/cron/test_cron_service.py b/tests/cron/test_cron_service.py index 185071f3d..19739d4cd 100644 --- a/tests/cron/test_cron_service.py +++ b/tests/cron/test_cron_service.py @@ -1,6 +1,7 @@ import asyncio import json import time +from pathlib import Path import pytest @@ -960,8 +961,8 @@ def test_stale_instance_remove_preserves_external_add(tmp_path) -> None: @pytest.mark.asyncio -async def test_save_store_failure_does_not_kill_scheduler(tmp_path, monkeypatch): - """A persistence failure in _on_timer must not stop future ticks.""" +async def test_save_store_failure_retries_without_replaying_job(tmp_path, monkeypatch): + """A failed post-run save must be retried before jobs can execute again.""" store_path = tmp_path / "cron" / "jobs.json" calls: list[str] = [] arm_calls: list[str] = [] @@ -990,25 +991,51 @@ async def test_save_store_failure_does_not_kill_scheduler(tmp_path, monkeypatch) service._save_store() arm_calls.clear() - # Simulate a disk-write failure on the next tick. - def failing_save() -> None: - raise OSError("disk full") + real_atomic_write = service._atomic_write + save_attempts = 0 + writes_fail = True - monkeypatch.setattr(service, "_save_store", failing_save) + def flaky_atomic_write(path: Path, content: str) -> None: + nonlocal save_attempts, writes_fail + save_attempts += 1 + if writes_fail: + raise OSError("disk full") + real_atomic_write(path, content) + + monkeypatch.setattr(service, "_atomic_write", flaky_atomic_write) await service._on_timer() - # The scheduler must still be re-armed after the failed save... + # The failed tick stays alive and retains the advanced in-memory state, + # including when a public read would normally reload from disk. assert arm_calls == ["arm"], "scheduler must re-arm after a failed tick" assert service._active_executions == 0 - # ...the first tick already ran the due job... assert calls == [job.id] - # ...and a later, healthy tick must still run the job again. - monkeypatch.setattr(service, "_save_store", CronService._save_store.__get__(service)) - job.state.next_run_at_ms = max(1, int(time.time() * 1000) - 1_000) + assert service._store_dirty is True + loaded = service.get_job(job.id) + assert loaded is not None + assert loaded.state.last_run_at_ms is not None + + # Manual execution is a second side-effecting entrypoint. It must also + # refuse to run until the previous result can be made durable. + with pytest.raises(OSError, match="disk full"): + await service.run_job(job.id, force=True) + assert calls == [job.id] + + # The next healthy tick is reserved for persisting the dirty snapshot. It + # must not reload the stale due record or execute the side effect twice. + writes_fail = False await service._on_timer() - assert calls == [job.id, job.id] - assert arm_calls == ["arm", "arm"] + assert calls == [job.id] + assert arm_calls == ["arm", "arm", "arm"] + assert save_attempts == 3 + assert service._store_dirty is False + + persisted = CronService(store_path).get_job(job.id) + assert persisted is not None + assert persisted.state.last_run_at_ms is not None + assert persisted.state.next_run_at_ms is not None + assert persisted.state.next_run_at_ms > persisted.state.last_run_at_ms @pytest.mark.asyncio