fix(cron): prevent replay after persistence failure

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:

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