mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-09-04 02:01:48 +03:00
fix(cron): keep scheduler alive when job-store persistence fails
A single OSError from _save_store() (disk full, permission change, locked file) escaped _on_timer's try/finally and killed the asyncio timer task, because _arm_timer() sits outside the block. All scheduled jobs silently stopped until restart or a manual re-arm via add_job/update_job/remove_job. Move _arm_timer() into the finally block and guard the whole tick body (including _load_store, which can persist during agent-binding migrations) so a transient persistence failure is logged and retried on the next tick instead of killing the scheduler. Add test_save_store_failure_does_not_kill_scheduler to cover the failure path that existing tests (which mock _arm_timer) never exercised.
This commit is contained in:
@@ -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."""
|
||||
|
||||
Reference in New Issue
Block a user