diff --git a/cron/jobs.py b/cron/jobs.py index 22d1b945b..1bccd6889 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -87,6 +87,16 @@ _jobs_lock_state = threading.local() OUTPUT_DIR = CRON_DIR / "output" ONESHOT_GRACE_SECONDS = 120 +# How long a one-shot's running-claim (#59229) is honored before it is +# considered stale and the job may be re-dispatched. The claim's real job is +# to be cleared by mark_job_run() the moment the run completes (success or +# failure); this TTL is only a safety valve for a claiming tick that DIED +# mid-run (gateway kill, OOM, hard-timeout) so a one-shot is never wedged +# forever. It must exceed the longest legitimate run: the default cron +# inactivity timeout is 600s and a job that keeps producing output can run +# past that, so 30 min gives generous headroom over any healthy run. +ONESHOT_RUN_CLAIM_TTL_SECONDS = 1800 + def _jobs_lock_file() -> Path: """Return the advisory lock path for the current cron directory.""" @@ -1306,6 +1316,11 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None, # Clear any external-fire claim so a re-armed recurring job can # be claimed again on its next fire (Phase 4C CAS). job["fire_claim"] = None + # Clear the one-shot running-claim (#59229): the run is over, so + # a re-armed recurring job or a re-dispatched one-shot recovery + # is claimable again. No-op if the job never carried a claim. + if job.get("run_claim") is not None: + job["run_claim"] = None # Increment completed count. Finite one-shot jobs are # pre-claimed by claim_dispatch() BEFORE the side effect runs @@ -1559,6 +1574,24 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]: if not job.get("enabled", True): continue + # Cross-process running-claim guard (#59229): if another scheduler + # process already claimed this one-shot and its run is still in flight + # (claim younger than the TTL), skip it — do NOT re-dispatch. The + # claim is stamped just before we return the job as due (below) and + # cleared by mark_job_run() on completion. A claim older than the TTL + # is treated as stale (the claiming tick died mid-run) and allowed + # through so the job is recovered rather than wedged forever. + existing_claim = job.get("run_claim") + if existing_claim and job.get("schedule", {}).get("kind") == "once": + try: + claimed_at = _ensure_aware( + datetime.fromisoformat(existing_claim["at"]) + ) + if (now - claimed_at).total_seconds() < ONESHOT_RUN_CLAIM_TTL_SECONDS: + continue # a fresh claim is held by an in-flight run + except (KeyError, ValueError, TypeError): + pass # malformed claim → fall through and (re)claim + next_run = job.get("next_run_at") if not next_run: schedule = job.get("schedule", {}) @@ -1703,20 +1736,27 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]: break continue - # Before returning a one-shot job as due, advance its - # next_run_at past the next scheduler tick so other processes - # (gateway + desktop running concurrent schedulers) don't - # double-execute it (#59229). The advance is persisted - # immediately under the same lock that get_due_jobs holds. - # mark_job_run re-anchors next_run_at on completion (success - # or failure), so a tick death between here and execution - # only delays the job by a single tick window — not lost. + # Durably claim a one-shot for the DURATION of its run before + # returning it as due, so a second scheduler process (gateway + + # desktop both run in-process 60s tickers on one HERMES_HOME) + # cannot re-dispatch it while the first run is still in flight + # (#59229). A plain one-shot's due-state is not resolved until + # mark_job_run() completes it minutes later, so advancing + # next_run_at by a fixed window is not enough — a job that outlives + # one tick (e.g. a 2.5-min research prompt) would simply re-fire on + # the next tick after the window. Instead we stamp a run_claim under + # the same lock get_due_jobs already holds; the other process reads + # a fresh claim on its next tick and skips (handled at the top of + # this loop). mark_job_run() clears the claim on completion. The TTL + # is only a safety valve: a claiming tick that DIES mid-run leaves a + # stale claim that expires after ONESHOT_RUN_CLAIM_TTL_SECONDS, so + # the job is re-dispatched rather than wedged forever. if kind == "once": - claimed_next = (now + timedelta(seconds=60)).isoformat() - job["next_run_at"] = claimed_next + claim = {"at": now.isoformat(), "by": _machine_id()} + job["run_claim"] = claim for rj in raw_jobs: if rj["id"] == job["id"]: - rj["next_run_at"] = claimed_next + rj["run_claim"] = claim needs_save = True break diff --git a/tests/cron/test_jobs.py b/tests/cron/test_jobs.py index 19080adf3..c34974689 100644 --- a/tests/cron/test_jobs.py +++ b/tests/cron/test_jobs.py @@ -927,9 +927,13 @@ class TestGetDueJobs: due = get_due_jobs() assert [job["id"] for job in due] == ["oneshot-recover"] - # next_run_at is now advanced past the tick window to prevent - # cross-process double-execution (#59229). - assert get_job("oneshot-recover")["next_run_at"] == (now + timedelta(seconds=60)).isoformat() + # Recovery restores next_run_at to the original run time; the + # cross-process double-exec guard (#59229) is a separate run_claim + # stamped under the lock, not a next_run_at mutation. + recovered = get_job("oneshot-recover") + assert recovered["next_run_at"] == run_at + assert recovered.get("run_claim") is not None + assert recovered["run_claim"]["at"] == now.isoformat() def test_broken_stale_one_shot_without_next_run_is_not_recovered(self, tmp_cron_dir, monkeypatch): now = datetime(2026, 3, 18, 4, 30, 0, tzinfo=timezone.utc) @@ -960,6 +964,80 @@ class TestGetDueJobs: assert get_due_jobs() == [] assert get_job("oneshot-stale")["next_run_at"] is None + def test_one_shot_not_redispatched_while_running(self, tmp_cron_dir, monkeypatch): + """#59229: two concurrent schedulers must not double-execute a one-shot. + + Reproduces the reported failure with a job whose run OUTLIVES the tick + interval (a ~2.5-min research prompt). Process A's tick returns it as + due and stamps a run_claim; while A is still running, every later tick + (process B, or A's own next tick) must see the fresh claim and skip — + not just for one tick window but for the whole run. + """ + from cron.jobs import _hermes_now + t0 = _hermes_now() + run_at = (t0 - timedelta(seconds=5)).isoformat() + save_jobs([{ + "id": "long-oneshot", "name": "R", "prompt": "2.5min research", + "schedule": {"kind": "once", "run_at": run_at}, + "next_run_at": run_at, "enabled": True, "state": "scheduled", + }]) + + # Process A tick: picks it up + claims it. + dueA = get_due_jobs() + assert [j["id"] for j in dueA] == ["long-oneshot"] + assert get_job("long-oneshot").get("run_claim") is not None + + # Process B (and A's own subsequent ticks) while A is still running: + # 28s later (the exact gap in the report) AND 61s later (past any + # fixed +60s window) — both must skip. + for gap in (28, 61, 130): + monkeypatch.setattr("cron.jobs._hermes_now", + lambda t0=t0, g=gap: t0 + timedelta(seconds=g)) + assert get_due_jobs() == [], f"double-dispatched at +{gap}s" + + def test_one_shot_run_claim_expires_after_ttl(self, tmp_cron_dir, monkeypatch): + """A claiming tick that DIED mid-run must not wedge the one-shot forever: + once the run_claim is older than the TTL it is re-dispatched (recovered).""" + from cron.jobs import _hermes_now, ONESHOT_RUN_CLAIM_TTL_SECONDS + t0 = _hermes_now() + run_at = (t0 - timedelta(seconds=5)).isoformat() + save_jobs([{ + "id": "wedged", "name": "R", "prompt": "x", + "schedule": {"kind": "once", "run_at": run_at}, + "next_run_at": run_at, "enabled": True, "state": "scheduled", + }]) + assert [j["id"] for j in get_due_jobs()] == ["wedged"] # A claims, then dies + + # Just inside the TTL: still claimed → skipped. + monkeypatch.setattr("cron.jobs._hermes_now", + lambda: t0 + timedelta(seconds=ONESHOT_RUN_CLAIM_TTL_SECONDS - 10)) + assert get_due_jobs() == [] + + # Just past the TTL: stale claim → re-dispatched (recovered), re-claimed. + monkeypatch.setattr("cron.jobs._hermes_now", + lambda: t0 + timedelta(seconds=ONESHOT_RUN_CLAIM_TTL_SECONDS + 10)) + recovered = get_due_jobs() + assert [j["id"] for j in recovered] == ["wedged"] + + def test_mark_job_run_clears_one_shot_run_claim(self, tmp_cron_dir, monkeypatch): + """mark_job_run() clears the run_claim on completion so a re-dispatched + one-shot (e.g. a stale-recovered retry) is claimable again.""" + from cron.jobs import _hermes_now + t0 = _hermes_now() + run_at = (t0 - timedelta(seconds=5)).isoformat() + # Give it repeat headroom so mark_job_run keeps the job around. + save_jobs([{ + "id": "claimclear", "name": "R", "prompt": "x", + "schedule": {"kind": "once", "run_at": run_at}, + "next_run_at": run_at, "enabled": True, "state": "scheduled", + "repeat": {"times": 2, "completed": 0}, + }]) + assert [j["id"] for j in get_due_jobs()] == ["claimclear"] + assert get_job("claimclear").get("run_claim") is not None + mark_job_run("claimclear", True) + assert get_job("claimclear")["run_claim"] is None + + def test_broken_cron_without_next_run_is_recovered(self, tmp_cron_dir, monkeypatch): now = datetime(2026, 3, 18, 10, 0, 0, tzinfo=timezone.utc) monkeypatch.setattr("cron.jobs._hermes_now", lambda: now)