diff --git a/nanobot/cron/service.py b/nanobot/cron/service.py index 20fbe4be6..b505bea2a 100644 --- a/nanobot/cron/service.py +++ b/nanobot/cron/service.py @@ -515,11 +515,10 @@ class CronService: self._active_executions += 1 try: 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 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: - self._arm_timer() return now = _now_ms() @@ -532,9 +531,21 @@ class CronService: await self._execute_job(job) self._save_store() + except Exception: + # A load/persist failure must not kill the scheduler: keep the + # in-memory store and retry on the next tick. This mirrors the + # read-path defense in ``_load_jobs`` (``.corrupt-`` backups); + # ``_load_store`` may also persist (agent-binding migrations). + logger.exception( + "Cron: tick failed ({}); " + "keeping in-memory state and retrying on next tick", + self.store_path, + ) finally: self._active_executions -= 1 - self._arm_timer() + # Always re-arm the timer, even on unexpected failures, so a + # single bad tick cannot silently stop all future jobs. + self._arm_timer() async def _execute_job(self, job: CronJob) -> None: """Execute a single job.""" diff --git a/tests/cron/test_cron_service.py b/tests/cron/test_cron_service.py index 5e18c6450..185071f3d 100644 --- a/tests/cron/test_cron_service.py +++ b/tests/cron/test_cron_service.py @@ -959,6 +959,58 @@ def test_stale_instance_remove_preserves_external_add(tmp_path) -> None: # ── timer race regression tests ── +@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.""" + store_path = tmp_path / "cron" / "jobs.json" + calls: list[str] = [] + arm_calls: list[str] = [] + + async def on_job(job): + calls.append(job.id) + + service = CronService(store_path, on_job=on_job) + service._running = True + service._load_store() + + # Spy on _arm_timer so we can assert the scheduler is re-armed even when + # the tick fails, without actually scheduling a real timer task. + def arm_spy() -> None: + arm_calls.append("arm") + + monkeypatch.setattr(service, "_arm_timer", arm_spy) + + job = service.add_job( + name="persist-failure", + schedule=CronSchedule(kind="every", every_ms=60_000), + message="hello", + **_bound_chat(), + ) + job.state.next_run_at_ms = max(1, int(time.time() * 1000) - 1_000) + service._save_store() + arm_calls.clear() + + # Simulate a disk-write failure on the next tick. + def failing_save() -> None: + raise OSError("disk full") + + monkeypatch.setattr(service, "_save_store", failing_save) + await service._on_timer() + + # The scheduler must still be re-armed after the failed save... + 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) + await service._on_timer() + + assert calls == [job.id, job.id] + assert arm_calls == ["arm", "arm"] + + @pytest.mark.asyncio async def test_timer_execution_is_not_rolled_back_by_list_jobs_reload(tmp_path): """list_jobs() during _on_timer should not replace the active store and re-run the same due job."""