From 261aef5268e1335d7e1740c2bb17045f8a4ae449 Mon Sep 17 00:00:00 2001 From: kshitij Date: Fri, 7 Aug 2026 17:09:14 +0530 Subject: [PATCH] perf(cron): stat-stamp fast path for the shrink-merge; no caller-list mutation Follow-up hardening on the salvaged #80687 shrink-merge guard, folding in the best part of the competing #80703 (credit: @JoaoMarcos44): - Stat-stamp fast path: load_jobs() inside a _jobs_lock() section records jobs.json's (mtime_ns, size, ino) BEFORE reading; the save-path merge and the post-stage verify skip their full read+parse when the stamp still matches. The healthy no-race save (every mark_job_run / claim_dispatch / heartbeat / advance_next_runs tick persist) now costs one stat() instead of up to two full JSON parses. - Fail-safe stamp discipline: the stamp is captured pre-read (a sibling racing the load leaves it older than disk -> mismatch -> merge runs), includes st_ino (mkstemp+rename always allocates a new inode, so coarse-mtime filesystems cannot false-match), resets on section entry/exit, and is INVALIDATED - never refreshed - after any write in the section (a refresh would let a nested create_job be clobbered by an outer caller's stale payload; probe-verified both directions). - _merge_unexpected_disk_jobs no longer mutates the caller's list in place - it returns a new list when anything was recovered, and logs the recovered ids. - The tolerant read cascade (utf-8-sig + strict=False fallback) is factored into one shared _parse_jobs_file used by both load_jobs and _peek_jobs_unlocked, so future encoding/shape fixes land once. The peek's repair-free re-entrancy contract is now documented - a repairing read on the save path would recurse through _save_jobs_unlocked (the exact defect the stamp-reconcile approach in #80703 had). 4 new regression tests (fast path, no-mutation, corrupt-file save, nested-create-vs-stale-outer-save), each verified to fail against the implementation it guards. Co-authored-by: joaomarcos --- cron/jobs.py | 163 ++++++++++++++++----- tests/cron/test_jobs_shrink_merge_80624.py | 100 +++++++++++++ 2 files changed, 223 insertions(+), 40 deletions(-) diff --git a/cron/jobs.py b/cron/jobs.py index 155b411522a5a..808241c8586f9 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -298,6 +298,12 @@ def _jobs_lock(): with _jobs_file_lock: _jobs_lock_state.depth = 1 + # Stamp of jobs.json as of this section's load_jobs() (#80703's + # fast-path, credit @JoaoMarcos44): lets _save_jobs_unlocked skip the + # shrink-merge parse when the file provably hasn't changed since this + # section read it. Reset on entry/exit so stale stamps from unlocked + # loads or prior sections can never suppress a needed merge. + _jobs_lock_state.load_stamp = None lock_fd = None try: try: @@ -363,6 +369,7 @@ def _jobs_lock(): lock_fd.close() finally: _jobs_lock_state.depth = 0 + _jobs_lock_state.load_stamp = None # Fields on a cron job that must never change after creation. ``id`` is used # as a filesystem path component under ``OUTPUT_DIR``; allowing it to be @@ -1010,33 +1017,43 @@ def get_ticker_last_error() -> Optional[str]: # Job CRUD Operations # ============================================================================= +def _parse_jobs_file(jobs_file: Path) -> Tuple[Any, bool]: + """Tolerantly parse jobs.json; shared by load_jobs and the save-path peek. + + Returns ``(data, used_strict_fallback)``. utf-8-sig absorbs a Windows + BOM; a strict parse failure is retried with ``strict=False`` to survive + bare control characters in string values. IO errors from the open and + parse errors from the fallback propagate to the caller, which decides + between repair (load_jobs) and bail-out (peek). + """ + with open(jobs_file, "r", encoding="utf-8-sig") as f: + raw = f.read() + try: + return json.loads(raw), False + except json.JSONDecodeError: + return json.loads(raw, strict=False), True + + def load_jobs() -> List[Dict[str, Any]]: """Load all jobs from storage.""" jobs_file = _current_cron_store().jobs_file ensure_dirs() + # Stamp BEFORE reading (fail-safe direction — see _record_load_stamp): + # a sibling write racing this load leaves the stamp older than disk, so + # the save-path merge runs instead of being wrongly skipped. + pre_read_stamp = _jobs_file_stamp(jobs_file) if not jobs_file.exists(): + _record_load_stamp(None) return [] - _strict_retry = False # track whether we used the strict=False fallback - try: - # utf-8-sig: Windows Notepad / PowerShell 5.1 Set-Content -Encoding UTF8 - # write a leading BOM; json.load under plain utf-8 raises - # JSONDecodeError("Unexpected UTF-8 BOM") and takes down cron. - with open(jobs_file, 'r', encoding='utf-8-sig') as f: - data = json.load(f) - except json.JSONDecodeError: - # Retry with strict=False to handle bare control chars in string values - _strict_retry = True - try: - with open(jobs_file, 'r', encoding='utf-8-sig') as f: - data = json.loads(f.read(), strict=False) - except Exception as e: - logger.error("Failed to auto-repair jobs.json: %s", e) - raise RuntimeError(f"Cron database corrupted and unrepairable: {e}") from e + data, _strict_retry = _parse_jobs_file(jobs_file) except IOError as e: logger.error("IOError reading jobs.json: %s", e) raise RuntimeError(f"Failed to read cron database: {e}") from e + except Exception as e: + logger.error("Failed to auto-repair jobs.json: %s", e) + raise RuntimeError(f"Cron database corrupted and unrepairable: {e}") from e # Validate the top-level JSON shape: accept a dict (expected) or a bare # list (auto-repair). Anything else (str/number/null) is corruption that @@ -1048,6 +1065,7 @@ def load_jobs() -> List[Dict[str, Any]]: # Hit control-character corruption — rewrite with proper escaping. save_jobs(jobs) logger.warning("Auto-repaired jobs.json (had invalid control characters)") + _record_load_stamp(pre_read_stamp) return jobs if isinstance(data, list): # Bare array — likely saved/edited outside save_jobs(). Wrap it back @@ -1055,6 +1073,7 @@ def load_jobs() -> List[Dict[str, Any]]: if data: save_jobs(data) logger.warning("Auto-repaired jobs.json (bare list wrapped as dict)") + _record_load_stamp(pre_read_stamp) return data raise RuntimeError( @@ -1067,20 +1086,18 @@ def _peek_jobs_unlocked() -> Optional[List[Dict[str, Any]]]: Caller must hold ``_jobs_lock()``. Returns ``[]`` when the file is missing, ``None`` when the payload is unreadable/corrupt (caller should - not attempt a shrink-merge against an unknown baseline). + not attempt a shrink-merge against an unknown baseline). Never calls + ``save_jobs`` — the repair-free property is what keeps the save path + re-entrancy-safe (a repairing read here would recurse through + ``_save_jobs_unlocked``). """ jobs_file = _current_cron_store().jobs_file if not jobs_file.exists(): return [] try: - with open(jobs_file, "r", encoding="utf-8-sig") as f: - data = json.load(f) + data, _ = _parse_jobs_file(jobs_file) except Exception: - try: - with open(jobs_file, "r", encoding="utf-8-sig") as f: - data = json.loads(f.read(), strict=False) - except Exception: - return None + return None if isinstance(data, dict): jobs = data.get("jobs", []) return jobs if isinstance(jobs, list) else None @@ -1089,12 +1106,48 @@ def _peek_jobs_unlocked() -> Optional[List[Dict[str, Any]]]: return None +def _jobs_file_stamp(jobs_file: Path) -> Optional[Tuple[int, int, int]]: + """Cheap change-detection stamp for jobs.json: ``(mtime_ns, size, ino)``. + + ``None`` means the file is missing/unstatable. Used as a fast-path gate + in front of the shrink-merge so the healthy no-race save costs one + ``stat()`` instead of a full read+parse (the ``advance_next_runs`` + batching exists because this path is hot — see its docstring). + ``st_ino`` is included because every legitimate writer goes through + mkstemp+rename (new inode), so even a same-size write inside one mtime + quantum on a coarse-clock filesystem (ext4 jiffies, network mounts) + cannot false-match. + """ + try: + st = jobs_file.stat() + return (st.st_mtime_ns, st.st_size, st.st_ino) + except OSError: + return None + + +def _record_load_stamp(stamp: Optional[Tuple[int, int, int]]) -> None: + """Remember jobs.json's stamp for the enclosing _jobs_lock() section. + + No-op outside a critical section. Lets the save path skip the + shrink-merge parse when the file provably hasn't changed since this + section loaded it (#80703's fast-path). The caller must capture the + stamp BEFORE reading the file: a sibling landing mid-read then leaves + the recorded stamp OLDER than disk — a mismatch, so the merge runs + (fail-safe direction). Stamping after the read would let that sibling's + write be certified as "seen" without being in the loaded payload, + wrongly suppressing the recovery. + """ + if not getattr(_jobs_lock_state, "depth", 0): + return + _jobs_lock_state.load_stamp = stamp + + def _merge_unexpected_disk_jobs( jobs: List[Dict[str, Any]], *, removed_ids: Optional[Collection[str]] = None, ) -> List[Dict[str, Any]]: - """Re-attach on-disk jobs missing from a save payload (#80624). + """Return *jobs* plus any on-disk jobs missing from the save payload (#80624). Under ``_jobs_lock()``'s degraded flock-timeout path (#60703), two processes can both believe they own the store. A writer that loaded an @@ -1105,8 +1158,18 @@ def _merge_unexpected_disk_jobs( Intentional deletes pass ``removed_ids``. Any other id present on disk but absent from *jobs* is treated as a concurrent create and merged - back before the atomic write. + back before the atomic write. The caller's list is never mutated — a + new list is returned when anything was recovered. + + Fast path: when the enclosing critical section recorded a load stamp + and the file's ``(mtime_ns, size)`` still matches, nothing can have + changed underneath us, so the read+parse is skipped entirely — one + ``stat()`` on the healthy no-race save. """ + stamp = getattr(_jobs_lock_state, "load_stamp", None) + if stamp is not None and _jobs_file_stamp(_current_cron_store().jobs_file) == stamp: + return jobs + disk_jobs = _peek_jobs_unlocked() if disk_jobs is None: return jobs @@ -1117,7 +1180,7 @@ def _merge_unexpected_disk_jobs( if isinstance(job, dict) and job.get("id"): new_ids.add(str(job["id"])) - preserved = 0 + recovered: List[Dict[str, Any]] = [] for disk_job in disk_jobs: if not isinstance(disk_job, dict): continue @@ -1127,18 +1190,19 @@ def _merge_unexpected_disk_jobs( disk_id = str(disk_id) if disk_id in new_ids or disk_id in intended_remove: continue - jobs.append(disk_job) + recovered.append(disk_job) new_ids.add(disk_id) - preserved += 1 - if preserved: - logger.warning( - "Preserved %d cron job(s) present on disk but missing from the " - "in-memory save payload (concurrent create under degraded lock " - "or stale writer) (#80624)", - preserved, - ) - return jobs + if not recovered: + return jobs + logger.warning( + "Preserved %d cron job(s) present on disk but missing from the " + "in-memory save payload (concurrent create under degraded lock " + "or stale writer) (#80624): %s", + len(recovered), + [j.get("id") for j in recovered], + ) + return jobs + recovered def _save_jobs_unlocked( @@ -1173,11 +1237,13 @@ def _save_jobs_unlocked( # path another process can create a job between our load and our write. # Merge unexpected disk ids into the payload, stage the write, then # re-peek; if new ids appeared, merge again and restage before replace. + # The merge itself fast-paths to a single stat() when the enclosing + # section's load stamp still matches (see _merge_unexpected_disk_jobs). tmp_path = None try: for _attempt in range(5): if not replace: - _merge_unexpected_disk_jobs(jobs, removed_ids=removed_ids) + jobs = _merge_unexpected_disk_jobs(jobs, removed_ids=removed_ids) fd, tmp_path = tempfile.mkstemp( dir=str(jobs_file.parent), suffix=".tmp", prefix=".jobs_" ) @@ -1199,7 +1265,15 @@ def _save_jobs_unlocked( raise if not replace: - disk_jobs = _peek_jobs_unlocked() + # Verify-after-stage: a sibling landing while we serialized + # the payload must trigger another merge round. Same stamp + # fast path as the merge — an unchanged stamp proves nothing + # was written, so the full parse is skipped. + _stamp = getattr(_jobs_lock_state, "load_stamp", None) + _unchanged = ( + _stamp is not None and _jobs_file_stamp(jobs_file) == _stamp + ) + disk_jobs = None if _unchanged else _peek_jobs_unlocked() if disk_jobs is not None: payload_ids = { str(j["id"]) @@ -1225,11 +1299,20 @@ def _save_jobs_unlocked( tmp_path = None _secure_file(jobs_file) _preserve_file_ownership(jobs_file, _stat_before) + # Invalidate (never refresh) the stamp after writing: the stamp + # certifies "this section's loaded payload still matches disk", + # which stops being provable the moment anyone writes. A refresh + # here would let a nested save (e.g. create_job inside a broader + # section) certify disk against an OUTER caller's stale payload + # and deterministically clobber the nested create; it also races + # a degraded sibling landing between replace and stat. Later + # saves in this section simply take the full merge (fail-safe). + _record_load_stamp(None) return # Exhausted retries — last merge + write without another re-peek. if not replace: - _merge_unexpected_disk_jobs(jobs, removed_ids=removed_ids) + jobs = _merge_unexpected_disk_jobs(jobs, removed_ids=removed_ids) fd, tmp_path = tempfile.mkstemp( dir=str(jobs_file.parent), suffix=".tmp", prefix=".jobs_" ) diff --git a/tests/cron/test_jobs_shrink_merge_80624.py b/tests/cron/test_jobs_shrink_merge_80624.py index 783d85fe6930c..423c3fe93d337 100644 --- a/tests/cron/test_jobs_shrink_merge_80624.py +++ b/tests/cron/test_jobs_shrink_merge_80624.py @@ -135,3 +135,103 @@ def test_jobs_json_on_disk_matches_merge(hermes_env): save_jobs([]) payload = json.loads((Path(hermes_env) / "cron" / "jobs.json").read_text()) assert [j["id"] for j in payload["jobs"]] == [job["id"]] + + +def test_stamp_fast_path_skips_merge_when_file_unchanged(hermes_env, monkeypatch): + """Inside a critical section whose load stamp still matches, the save + must not re-read jobs.json at all (#80703's single-stat fast path).""" + import cron.jobs as jobs + from cron.jobs import create_job + + job = create_job( + prompt=None, + schedule="every 2m", + script="watch.sh", + no_agent=True, + deliver="local", + name="watchdog", + repeat=0, + ) + + peeks = {"count": 0} + real_peek = jobs._peek_jobs_unlocked + + def _counting_peek(): + peeks["count"] += 1 + return real_peek() + + monkeypatch.setattr(jobs, "_peek_jobs_unlocked", _counting_peek) + + # load -> save inside ONE critical section, no sibling write in between: + # the stamp matches, so the merge (and its peek) must be skipped. + with jobs._jobs_lock(): + current = jobs.load_jobs() + jobs._save_jobs_unlocked(current) + assert peeks["count"] == 0, "healthy same-section save should not re-parse" + + # A sibling write invalidates the stamp -> the merge runs again. + with jobs._jobs_lock(): + current = jobs.load_jobs() + (jobs._current_cron_store().jobs_file).write_text( + json.dumps({"jobs": [dict(job), {"id": "bbbbbbbbbbbb", "name": "b"}]}), + encoding="utf-8", + ) + jobs._save_jobs_unlocked(current) + assert peeks["count"] > 0, "changed stamp must re-trigger the merge" + ids = {j["id"] for j in jobs.load_jobs()} + assert ids == {job["id"], "bbbbbbbbbbbb"} + + +def test_merge_does_not_mutate_caller_list(hermes_env): + """The shrink-merge returns a new list; the caller's payload object must + not grow as a side effect of save_jobs().""" + from cron.jobs import create_job, save_jobs + + job = create_job( + prompt=None, + schedule="every 2m", + script="watch.sh", + no_agent=True, + deliver="local", + name="watchdog", + repeat=0, + ) + my_payload = [] # stale snapshot from before the create + save_jobs(my_payload) + assert my_payload == [], "caller's list was mutated in place by the merge" + + +def test_corrupt_disk_file_does_not_break_save(hermes_env): + """A corrupt jobs.json under a save must not recurse or crash: the + non-repairing peek returns None and the save overwrites cleanly.""" + import cron.jobs as jobs + from cron.jobs import load_jobs, save_jobs + + jobs.ensure_dirs() + jobs_file = jobs._current_cron_store().jobs_file + jobs_file.write_text('{"jobs": [{"id": "ccc', encoding="utf-8") + save_jobs([{"id": "aaaaaaaaaaaa", "name": "a"}]) + assert [j["id"] for j in load_jobs()] == ["aaaaaaaaaaaa"] + + +def test_nested_create_survives_outer_stale_save(hermes_env): + """A save inside a critical section invalidates the section's stamp, so + an outer caller's later save with a pre-create payload must re-merge and + keep the nested create (stamp refresh here would deterministically + clobber it).""" + import cron.jobs as jobs + from cron.jobs import create_job, load_jobs, save_jobs + + seed = {"id": "aaaaaaaaaaaa", "name": "a"} + save_jobs([seed], replace=True) + + with jobs._jobs_lock(): + stale = jobs.load_jobs() # records the stamp + created = create_job( + prompt="x", schedule="every 5m", name="nested", deliver="local" + ) + jobs._save_jobs_unlocked(stale) # outer save with pre-create payload + + ids = {j["id"] for j in load_jobs()} + assert created["id"] in ids, "nested create was clobbered by outer stale save" + assert seed["id"] in ids