diff --git a/cron/jobs.py b/cron/jobs.py index 044d0bee54ecf..6c7c2b21ecf41 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -235,15 +235,18 @@ def _release_flock(lock_fd) -> None: @contextlib.contextmanager -def _jobs_lock(): +def _jobs_lock(*, strict: bool = False): """Serialize a load_jobs→modify→save_jobs critical section: in-process RLock (parallel tick threads) plus a cross-process flock on ``/.jobs.lock`` (gateway vs. CLI writes — otherwise a `cron pause` could be clobbered and keep firing). Nested calls in one thread reuse the held lock. Without a flock backend, or on flock timeout (logged loudly), it degrades to in-process-only locking: a briefly torn cross-process write beats a dead - scheduler.""" + scheduler. ``strict=True`` instead fails closed; nested strict calls require an actually + acquired outer flock, not merely an in-process lock.""" depth = getattr(_jobs_lock_state, "depth", 0) if depth: + if strict and not getattr(_jobs_lock_state, "flock_acquired", False): + raise RuntimeError("Cron jobs lock is degraded; strict locking required") _jobs_lock_state.depth = depth + 1 try: yield @@ -258,13 +261,20 @@ def _jobs_lock(): # stamps from unlocked loads or prior sections can never suppress a needed merge. # See #80703. _jobs_lock_state.load_stamp = None + _jobs_lock_state.flock_acquired = False lock_fd = None try: try: ensure_dirs() lock_fd = open(_jobs_lock_file(), "a+", encoding="utf-8") lock_fd.seek(0) - if _acquire_flock(lock_fd, _JOBS_LOCK_TIMEOUT_SECONDS) is False: + acquired = _acquire_flock(lock_fd, _JOBS_LOCK_TIMEOUT_SECONDS) + _jobs_lock_state.flock_acquired = acquired is True + if strict and not _jobs_lock_state.flock_acquired: + if acquired is False: + raise TimeoutError("Timed out waiting for the cron jobs lock") + raise RuntimeError("No cross-process lock backend for cron jobs lock") + if acquired is False: logger.error( "Timed out after %.0fs waiting for the cron " "jobs lock (%s) — another process is holding " @@ -275,17 +285,22 @@ def _jobs_lock(): lock_fd.close() lock_fd = None except (OSError, IOError) as e: + if strict: + raise # A locking failure must never take down cron writes — in-process lock still held. logger.warning("jobs.json cross-process lock unavailable (%s); " "proceeding with in-process lock only", e) + yield + finally: try: - yield - finally: - if lock_fd is not None: + if lock_fd is not None and _jobs_lock_state.flock_acquired: _release_flock(lock_fd) - finally: - _jobs_lock_state.depth = 0 - _jobs_lock_state.load_stamp = None + elif lock_fd is not None: + lock_fd.close() + finally: + _jobs_lock_state.depth = 0 + _jobs_lock_state.load_stamp = None + _jobs_lock_state.flock_acquired = False @contextlib.contextmanager @@ -2517,7 +2532,14 @@ def heartbeat_fire_claim(job_id: str, *, expected_owner: str) -> bool: def apply(jobs, _i, job): return _refresh_claim(jobs, job.get("fire_claim"), expected_owner) - return _under_fire_fence(job_id, lambda: _with_job(job_id, apply, False)) + # Renewal only changes the timestamp: the jobs lock makes the owner check + write atomic + # against takeover/completion, just like heartbeat_run_claim. Taking the fire fence here + # would block behind our own delivery and mistake its lock timeout for ownership loss. + # Keep that fence on owner mutations and external side effects, not lease renewal. + # Do not renew from an unsynchronized snapshot if the jobs lock degrades. Raising lets + # the heartbeat loop apply its existing transient-error grace instead of reporting loss. + with _jobs_lock(strict=True): + return _with_job(job_id, apply, False) # Completed one-shots are retained in jobs.json (final status stays inspectable) and pruned by diff --git a/tests/cron/test_heartbeat_delivery_fence.py b/tests/cron/test_heartbeat_delivery_fence.py new file mode 100644 index 0000000000000..964742308543c --- /dev/null +++ b/tests/cron/test_heartbeat_delivery_fence.py @@ -0,0 +1,122 @@ +"""Renewal must not contend with delivery, without weakening owner fencing.""" + +import threading + +import pytest + + +@pytest.mark.parametrize("claim_kind", ["fire", "run"]) +def test_claim_renewal_during_side_effect_fence(tmp_path, monkeypatch, claim_kind): + from cron import jobs + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr(jobs, "_JOBS_LOCK_TIMEOUT_SECONDS", 0.05) + job = jobs.create_job(prompt="x", schedule="in 30m") + claimed = jobs.claim_job_for_fire(job["id"], return_job=True) + owner = claimed["fire_claim"]["by"] + field = f"{claim_kind}_claim" + if claim_kind == "run": + jobs.update_job(job["id"], {field: dict(claimed["fire_claim"])}) + before = jobs.get_job(job["id"])[field] + heartbeat = getattr(jobs, f"heartbeat_{claim_kind}_claim") + results = [] + errors = [] + done = threading.Event() + + def renew(): + try: + results.append(heartbeat(job["id"], expected_owner=owner)) + # A competing dispatch and terminal write must still fail closed. + results.append(jobs.claim_job_for_fire(job["id"], claim_ttl_seconds=0)) + results.append(jobs.mark_job_run(job["id"], True, expected_fire_owner=owner)) + except BaseException as exc: + errors.append(exc) + finally: + done.set() + + with jobs.fire_claim_fence(job["id"], expected_owner=owner) as owned: + assert owned + thread = threading.Thread(target=renew) + thread.start() + try: + assert done.wait(5), "renewal blocked behind delivery's fire fence" + finally: + thread.join(timeout=5) + assert not thread.is_alive() + assert not errors + assert results == [True, False, False] + refreshed = jobs.get_job(job["id"])[field] + assert refreshed["by"] == before["by"] + assert refreshed["at"] != before["at"] + + # Replaced or cleared claims must never be renewed by the stale owner. + replacement = {"by": "replacement", "at": refreshed["at"]} + jobs.update_job(job["id"], {field: replacement}) + assert heartbeat(job["id"], expected_owner=owner) is False + assert jobs.get_job(job["id"])[field] == replacement + jobs.update_job(job["id"], {field: None}) + assert heartbeat(job["id"], expected_owner=owner) is False + assert jobs.get_job(job["id"])[field] is None + + +@pytest.mark.parametrize("replace_owner", [False, True]) +def test_run_one_job_delivery_and_real_ownership_loss(tmp_path, monkeypatch, replace_owner): + from cron import executions, jobs, scheduler + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr(jobs, "_JOBS_LOCK_TIMEOUT_SECONDS", 0.05) + monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.01) + job = jobs.create_job(prompt="x", schedule="every 5m", deliver="discord:123") + claimed = jobs.claim_job_for_fire(job["id"], return_job=True) + execution = executions.create_execution(job["id"], source="direct") + claimed["execution_id"] = execution["id"] + delivering = threading.Event() + renewed = threading.Event() + renewals = [] + delivered = [] + real_heartbeat = jobs.heartbeat_fire_claim + + def observe_heartbeat(job_id, *, expected_owner): + result = real_heartbeat(job_id, expected_owner=expected_owner) + if delivering.is_set(): + renewals.append(result) + renewed.set() + return result + + def run_job(_job, *, cancel_event, **kwargs): + if replace_owner: + assert jobs.claim_job_for_fire(job["id"], claim_ttl_seconds=0) + assert cancel_event.wait(5), "real replacement did not cancel the stale runner" + return True, "saved output", "final response", None + + def deliver(_job, content, **kwargs): + delivered.append(content) + delivering.set() + try: + # Delivery holds the real fire fence until the monitor has attempted renewal. + assert renewed.wait(5), "heartbeat never completed during delivery" + finally: + delivering.clear() + return None + + monkeypatch.setattr(scheduler, "heartbeat_fire_claim", observe_heartbeat) + monkeypatch.setattr(scheduler, "run_job", run_job) + monkeypatch.setattr(scheduler, "_deliver_result", deliver) + assert scheduler.run_one_job(claimed) is True + + persisted = jobs.get_job(job["id"]) + ledger = executions.get_execution(execution["id"]) + if replace_owner: + assert delivered == [] + assert ledger["status"] == "failed" + assert "ownership lost" in ledger["error"].lower() + assert persisted["fire_claim"]["by"] != claimed["fire_claim"]["by"] + assert persisted.get("last_run_at") is None + else: + assert delivered == ["final response"] + assert renewals and all(renewals) + assert persisted["last_status"] == "ok" + assert persisted["fire_claim"] is None + assert ledger["status"] == "completed" + assert ledger["error"] is None + assert list((tmp_path / "cron" / "output" / job["id"]).glob("*.md")) diff --git a/tests/cron/test_heartbeat_strict_lock.py b/tests/cron/test_heartbeat_strict_lock.py new file mode 100644 index 0000000000000..78a595e94f002 --- /dev/null +++ b/tests/cron/test_heartbeat_strict_lock.py @@ -0,0 +1,68 @@ +"""Lease renewal must never use the jobs lock's degraded write mode.""" +import contextlib +from datetime import datetime, timedelta +from unittest.mock import Mock + +import pytest + + +@pytest.mark.parametrize("nested", [False, True]) +@pytest.mark.parametrize("failure", ["timeout", "missing", "oserror"]) +def test_heartbeat_fails_closed_without_cross_process_lock(tmp_path, monkeypatch, failure, nested): + import cron.jobs as jobs + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + job = jobs.create_job(prompt="x", schedule="every 5m") + assert jobs.claim_job_for_fire(job["id"]) + claim = jobs.get_job(job["id"])["fire_claim"] + now = datetime.fromisoformat(claim["at"]) + timedelta(seconds=30) + monkeypatch.setattr(jobs, "_hermes_now", lambda: now) + before = (tmp_path / "cron" / "jobs.json").read_bytes() + + if failure == "missing": + monkeypatch.setattr(jobs, "fcntl", None) + monkeypatch.setattr(jobs, "msvcrt", None) + else: + acquire = ( + Mock(return_value=False) if failure == "timeout" + else Mock(side_effect=OSError("lock failed")) + ) + monkeypatch.setattr(jobs, "_acquire_flock", acquire) + load = Mock(wraps=jobs.load_jobs) + monkeypatch.setattr(jobs, "load_jobs", load) + + error = RuntimeError if nested or failure == "missing" else ( + TimeoutError if failure == "timeout" else OSError + ) + with jobs._jobs_lock() if nested else contextlib.nullcontext(): + with pytest.raises(error): + jobs.heartbeat_fire_claim(job["id"], expected_owner=claim["by"]) + load.assert_not_called() + assert (tmp_path / "cron" / "jobs.json").read_bytes() == before + # An unsuccessful acquisition must close its descriptor, including an OSError. + if failure != "missing": + assert all(call.args[0].closed for call in acquire.call_args_list) + assert jobs._jobs_lock_state.depth == 0 + + +@pytest.mark.parametrize("strict_outer", [False, True]) +def test_heartbeat_reuses_real_outer_lock_and_preserves_owner(tmp_path, monkeypatch, strict_outer): + import cron.jobs as jobs + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + job = jobs.create_job(prompt="x", schedule="every 5m") + assert jobs.claim_job_for_fire(job["id"]) + claim = jobs.get_job(job["id"])["fire_claim"] + now = datetime.fromisoformat(claim["at"]) + timedelta(seconds=30) + monkeypatch.setattr(jobs, "_hermes_now", lambda: now) + with jobs._jobs_lock(strict=strict_outer): + acquire = Mock(side_effect=AssertionError("nested lock must reuse actual flock")) + monkeypatch.setattr(jobs, "_acquire_flock", acquire) + assert jobs.heartbeat_fire_claim(job["id"], expected_owner=claim["by"]) + refreshed = jobs.get_job(job["id"])["fire_claim"] + assert refreshed["at"] != claim["at"] + assert refreshed["by"] == claim["by"] + assert not jobs.heartbeat_fire_claim(job["id"], expected_owner="stale-owner") + assert jobs.get_job(job["id"])["fire_claim"] == refreshed + acquire.assert_not_called() + assert jobs._jobs_lock_state.depth == 0