From 947310437b2548d75e4a4e384deaad068bf8ca47 Mon Sep 17 00:00:00 2001 From: spfcraze Date: Sat, 1 Aug 2026 11:59:39 -0400 Subject: [PATCH] perf(cron): batch advance_next_run for the due-dispatch loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The scheduler's pre-dispatch loop called advance_next_run per due job — one full load_jobs() + one full save_jobs() of the jobs file each — so N due jobs cost N reads + N writes of the whole file (gateway-restart catch-up or co-scheduled bursts). advance_next_runs() does one load + at most one save for the whole due set with identical per-job semantics; advance_next_run() is now a thin wrapper over it. Measured (50 due recurring jobs, real jobs file): 107.9 ms -> 2.5 ms (45x; 50 loads + 50 saves -> 1 + 1). Tests: batch advances recurring and skips one-shots, single load + save I/O pin (fails pre-fix — no such function), no save when nothing advances, and per-job wrapper semantics unchanged. Related: #60946 and #75833 both restructure this loop's call site for correctness — neither addresses the I/O cost, and this batch primitive composes with either dispatch design; happy to rebase onto whichever lands first. --- cron/jobs.py | 53 +++++++++++++------ cron/scheduler.py | 8 +-- tests/cron/test_execution_ledger.py | 2 +- tests/cron/test_jobs.py | 70 ++++++++++++++++++++++++++ tests/cron/test_parallel_pool.py | 45 +++++++++++++++-- tests/cron/test_run_one_job.py | 2 +- tests/cron/test_scheduler.py | 6 +-- tests/cron/test_sessiondb_init_hang.py | 2 +- 8 files changed, 160 insertions(+), 28 deletions(-) diff --git a/cron/jobs.py b/cron/jobs.py index e1b775bb4b505..479cf0bee5b5b 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -1923,6 +1923,43 @@ def heartbeat_run_claim(job_id: str, *, expected_owner: str) -> bool: return False +def advance_next_runs(job_ids) -> int: + """Batch form of :func:`advance_next_run` for the due-dispatch loop. + + One ``load_jobs()`` + at most one ``save_jobs()`` for the whole due + set, instead of one of each per job — the per-job form costs + O(N loads + N saves) for N due jobs (~110 ms at N=50, measured), the + batch form O(1 + 1) (~2 ms). ``job_ids`` may contain ids of one-shot + or unknown jobs; they are skipped exactly as the per-job form skips + them. Returns the number of jobs whose ``next_run_at`` was advanced. + + Crash semantics: the batch persists once at the end, so a crash + mid-batch re-fires the whole set on restart (at-least-once burst) + rather than advancing a prefix — acceptable given the sub-10ms window, + and identical to the per-job form once the batch completes. + """ + ids = set(job_ids) + if not ids: + return 0 + with _jobs_lock(): + jobs = load_jobs() + now = _hermes_now().isoformat() + advanced = 0 + for job in jobs: + if job["id"] not in ids: + continue + kind = job.get("schedule", {}).get("kind") + if kind not in {"cron", "interval"}: + continue + new_next = compute_next_run(job["schedule"], now) + if new_next and new_next != job.get("next_run_at"): + job["next_run_at"] = new_next + advanced += 1 + if advanced: + save_jobs(jobs) + return advanced + + def advance_next_run(job_id: str) -> bool: """Preemptively advance next_run_at for a recurring job before execution. @@ -1935,21 +1972,7 @@ def advance_next_run(job_id: str) -> bool: Returns True if next_run_at was advanced, False otherwise. """ - with _jobs_lock(): - jobs = load_jobs() - for job in jobs: - if job["id"] == job_id: - kind = job.get("schedule", {}).get("kind") - if kind not in {"cron", "interval"}: - return False - now = _hermes_now().isoformat() - new_next = compute_next_run(job["schedule"], now) - if new_next and new_next != job.get("next_run_at"): - job["next_run_at"] = new_next - save_jobs(jobs) - return True - return False - return False + return advance_next_runs([job_id]) == 1 def _machine_id() -> str: diff --git a/cron/scheduler.py b/cron/scheduler.py index 77c2772762238..431c24ffc805b 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -281,7 +281,7 @@ _LEGACY_HOME_TARGET_ENV_VARS = { "QQBOT_HOME_CHANNEL": "QQ_HOME_CHANNEL", } -from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_run, claim_dispatch, heartbeat_run_claim +from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_run, advance_next_runs, claim_dispatch, heartbeat_run_claim from cron.executions import create_execution, finish_execution, mark_execution_running # Sentinel: when a cron agent has nothing new to report, it can start its @@ -4153,11 +4153,11 @@ def tick( # Advance next_run_at for all recurring jobs FIRST, under the file lock, # before any execution begins. This preserves at-most-once semantics. - # For parallel jobs that are already running, advance_next_run keeps + # For parallel jobs that are already running, the advance keeps # bumping next_run_at forward so the grace window never expires. # mark_job_run() overwrites next_run_at on completion. - for job in due_jobs: - advance_next_run(job["id"]) + # Batched: one load + one save for the whole due set, not one per job. + advance_next_runs([job["id"] for job in due_jobs]) # Resolve max parallel workers: env var > config.yaml > unbounded. # Set HERMES_CRON_MAX_PARALLEL=1 to restore old serial behaviour. diff --git a/tests/cron/test_execution_ledger.py b/tests/cron/test_execution_ledger.py index 19c48e1cbd6e7..62478aa95be44 100644 --- a/tests/cron/test_execution_ledger.py +++ b/tests/cron/test_execution_ledger.py @@ -182,7 +182,7 @@ def test_generic_submit_failure_finishes_attempt_and_releases_guard(monkeypatch) lambda execution_id, **kwargs: finished.append((execution_id, kwargs)), ) monkeypatch.setattr(scheduler, "get_due_jobs", lambda: [{"id": "submit-fail"}]) - monkeypatch.setattr(scheduler, "advance_next_run", lambda _job_id: None) + monkeypatch.setattr(scheduler, "advance_next_runs", lambda _ids: 0) monkeypatch.setattr(scheduler, "_get_parallel_pool", lambda _workers: BrokenPool()) assert scheduler.tick(verbose=False, sync=False) == 0 diff --git a/tests/cron/test_jobs.py b/tests/cron/test_jobs.py index 6bfb7edffdf23..e9402208e3cc4 100644 --- a/tests/cron/test_jobs.py +++ b/tests/cron/test_jobs.py @@ -1069,3 +1069,73 @@ class TestJobsJsonUtf8Bom: assert [j["id"] for j in loaded] == ["plainjob01"] + + +class TestAdvanceNextRuns: + """Tests for advance_next_runs() — the batched due-set advance. + + The scheduler's pre-dispatch loop advanced each due job individually: + N due jobs = N full load_jobs() + N full save_jobs() of the jobs file + (~110 ms at N=50, measured). The batch form does one load + at most + one save (~2 ms). Imported inside test bodies so the pre-fix tree + fails with a real test failure (ImportError), not a collection error. + """ + + def _make_due(self, tmp_cron_dir, n_recurring=3, n_oneshot=1): + rec = [create_job(prompt=f"rec {i}", schedule="every 1h") + for i in range(n_recurring)] + one = [create_job(prompt=f"one {i}", schedule="30m") + for i in range(n_oneshot)] + jobs = load_jobs() + old = (datetime.now() - timedelta(minutes=5)).isoformat() + for j in jobs: + j["next_run_at"] = old + save_jobs(jobs) + return [j["id"] for j in rec], [j["id"] for j in one] + + def test_batch_advances_recurring_skips_oneshots(self, tmp_cron_dir): + from cron.jobs import advance_next_runs + rec_ids, one_ids = self._make_due(tmp_cron_dir) + advanced = advance_next_runs(rec_ids + one_ids) + assert advanced == len(rec_ids) + from cron.jobs import _ensure_aware, _hermes_now + for jid in rec_ids: + nxt = _ensure_aware(datetime.fromisoformat(get_job(jid)["next_run_at"])) + assert nxt > _hermes_now() + for jid in one_ids: + # one-shots keep their (past) next_run_at for restart retry + assert datetime.fromisoformat(get_job(jid)["next_run_at"]) < datetime.now() + + def test_batch_single_load_and_save(self, tmp_cron_dir, monkeypatch): + """I/O pin: the whole due set costs one load + one save, not N+N. + Fails pre-fix (function absent) and would fail on any regression + back to per-job I/O.""" + from cron.jobs import advance_next_runs + rec_ids, _ = self._make_due(tmp_cron_dir, n_recurring=10, n_oneshot=0) + import cron.jobs as cj + counts = {"load": 0, "save": 0} + real_load, real_save = cj.load_jobs, cj.save_jobs + monkeypatch.setattr(cj, "load_jobs", lambda *a, **k: ( + counts.__setitem__("load", counts["load"] + 1), real_load(*a, **k))[1]) + monkeypatch.setattr(cj, "save_jobs", lambda *a, **k: ( + counts.__setitem__("save", counts["save"] + 1), real_save(*a, **k))[1]) + advance_next_runs(rec_ids) + assert counts == {"load": 1, "save": 1} + + def test_batch_no_save_when_nothing_advances(self, tmp_cron_dir, monkeypatch): + from cron.jobs import advance_next_runs + rec_ids, one_ids = self._make_due(tmp_cron_dir, n_recurring=0, n_oneshot=2) + import cron.jobs as cj + saves = [0] + real_save = cj.save_jobs + monkeypatch.setattr(cj, "save_jobs", lambda *a, **k: ( + saves.__setitem__(0, saves[0] + 1), real_save(*a, **k))[1]) + assert advance_next_runs(one_ids + ["missing-id"]) == 0 + assert saves[0] == 0 + + def test_wrapper_semantics_unchanged(self, tmp_cron_dir): + """advance_next_run keeps its per-job contract over the batch.""" + rec_ids, one_ids = self._make_due(tmp_cron_dir) + assert advance_next_run(rec_ids[0]) is True + assert advance_next_run(one_ids[0]) is False + assert advance_next_run("missing-id") is False diff --git a/tests/cron/test_parallel_pool.py b/tests/cron/test_parallel_pool.py index 12903830cc75f..67ba0a32bcd7c 100644 --- a/tests/cron/test_parallel_pool.py +++ b/tests/cron/test_parallel_pool.py @@ -71,7 +71,7 @@ class TestRunningJobGuard: dispatched = [] monkeypatch.setattr(sched, "get_due_jobs", lambda: [job]) - monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "advance_next_runs", lambda *_a, **_kw: 0) monkeypatch.setattr(sched, "run_job", lambda j, **_kw: dispatched.append(j["id"]) or (True, "out", "resp", None)) monkeypatch.setattr(sched, "save_job_output", lambda *_a, **_kw: None) monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) @@ -104,7 +104,7 @@ class TestSyncMode: ] monkeypatch.setattr(sched, "get_due_jobs", lambda: jobs) - monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "advance_next_runs", lambda *_a, **_kw: 0) monkeypatch.setattr(sched, "run_job", lambda j, **_kw: (True, "out", "resp", None)) monkeypatch.setattr(sched, "save_job_output", lambda *_a, **_kw: "/tmp/out") monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) @@ -151,7 +151,7 @@ class TestSequentialPool: return True, "out", "resp", None monkeypatch.setattr(sched, "get_due_jobs", lambda: [job]) - monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "advance_next_runs", lambda *_a, **_kw: 0) monkeypatch.setattr(sched, "run_job", slow_run) monkeypatch.setattr(sched, "save_job_output", lambda *_a, **_kw: "/tmp/out") monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) @@ -180,3 +180,42 @@ class TestSequentialPool: sched._shutdown_parallel_pool() assert sched._sequential_pool is None + + +class TestTickBatchAdvance: + """The tick's pre-dispatch advance must go through advance_next_runs + exactly once with the whole due set — a revert to the per-job loop + (or back to advance_next_run) must fail this test, not slip past the + helper-level I/O pin.""" + + def test_tick_calls_advance_next_runs_once_with_all_due_ids(self, tmp_path, monkeypatch): + import cron.scheduler as sched + + sched._parallel_pool = None + sched._parallel_pool_max_workers = None + sched._running_job_ids.clear() + + jobs = [ + {"id": f"job-{i}", "name": f"Job {i}", "prompt": "test", + "schedule": "every 5m", "enabled": True, + "next_run_at": "2020-01-01T00:00:00", "deliver": "local"} + for i in range(4) + ] + + advance_calls = [] + monkeypatch.setattr(sched, "get_due_jobs", lambda: jobs) + monkeypatch.setattr( + sched, "advance_next_runs", + lambda ids: advance_calls.append(list(ids)) or len(list(ids))) + monkeypatch.setattr(sched, "run_job", lambda j, **_kw: (True, "out", "resp", None)) + monkeypatch.setattr(sched, "save_job_output", lambda *_a, **_kw: "/tmp/out") + monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "_deliver_result", lambda *_a, **_kw: None) + + n = sched.tick(verbose=False) + + assert n == 4 + assert advance_calls == [["job-0", "job-1", "job-2", "job-3"]], ( + f"tick must batch-advance the due set in ONE call; got {advance_calls}") + + sched._shutdown_parallel_pool() diff --git a/tests/cron/test_run_one_job.py b/tests/cron/test_run_one_job.py index c79d79cfa9849..d147d7e717ac5 100644 --- a/tests/cron/test_run_one_job.py +++ b/tests/cron/test_run_one_job.py @@ -46,7 +46,7 @@ def test_tick_process_job_sequence(monkeypatch): sequence run_job → save → deliver → mark, in that order.""" calls = _patch_pipeline(monkeypatch) monkeypatch.setattr(s, "get_due_jobs", lambda: [{"id": "j1", "name": "t"}]) - monkeypatch.setattr(s, "advance_next_run", lambda jid: True) + monkeypatch.setattr(s, "advance_next_runs", lambda ids: 1) s.tick(verbose=False, sync=True) diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 3b91c13579825..dd16c805a0e34 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -545,7 +545,7 @@ class TestRunJobSessionPersistence: "enabled": True, } with patch("cron.scheduler.get_due_jobs", return_value=[job]), patch( - "cron.scheduler.advance_next_run" + "cron.scheduler.advance_next_runs" ) as advance, patch("cron.scheduler.run_one_job") as run_one: assert tick(verbose=False, sync=True, can_dispatch=lambda: False) == 0 @@ -1230,7 +1230,7 @@ class TestParallelTick: ] with patch("cron.scheduler.get_due_jobs", return_value=jobs), \ - patch("cron.scheduler.advance_next_run"), \ + patch("cron.scheduler.advance_next_runs"), \ patch("cron.scheduler.run_job", side_effect=mock_run_job), \ patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ patch("cron.scheduler._deliver_result", return_value=None), \ @@ -1275,7 +1275,7 @@ class TestParallelTick: ] with patch("cron.scheduler.get_due_jobs", return_value=jobs), \ - patch("cron.scheduler.advance_next_run"), \ + patch("cron.scheduler.advance_next_runs"), \ patch("cron.scheduler.run_job", side_effect=mock_run_job), \ patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ patch("cron.scheduler._deliver_result", return_value=None), \ diff --git a/tests/cron/test_sessiondb_init_hang.py b/tests/cron/test_sessiondb_init_hang.py index e054773561ed6..9f89574309b8d 100644 --- a/tests/cron/test_sessiondb_init_hang.py +++ b/tests/cron/test_sessiondb_init_hang.py @@ -222,7 +222,7 @@ class TestDispatchGuardReleasedAfterHang: side_effect=_session_db_executor(timeouts), ), \ patch.object(sched, "get_due_jobs", return_value=[job]), \ - patch.object(sched, "advance_next_run"), \ + patch.object(sched, "advance_next_runs"), \ patch.object(sched, "save_job_output", return_value="/tmp/out"), \ patch.object(sched, "mark_job_run"), \ patch.object(sched, "_deliver_result", return_value=None):