diff --git a/cron/scheduler.py b/cron/scheduler.py index 9bffddc07259b..6c66fcdd862fb 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -1115,6 +1115,8 @@ def drain_delivery_queue(adapters, loop) -> int: _SCRIPT_TIMEOUT = _DEFAULT_SCRIPT_TIMEOUT _RUN_CLAIM_HEARTBEAT_SECONDS = 60.0 _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS = _RUN_CLAIM_HEARTBEAT_SECONDS * 3 +# Pause before re-sampling a fire-claim heartbeat miss; a genuinely re-owned claim misses twice. +_FIRE_CLAIM_MISS_CONFIRM_SECONDS = 1.0 def _cron_cleanup_timeout_seconds() -> float: @@ -2421,6 +2423,12 @@ def _heartbeat_loop() -> None: if self_removal_delivery_allowed(job_id): # Record dropped by this run; nothing left to keep fresh. continue + # One miss is a sample, not a verdict (#113357): a latch cancels the live + # agent run and ends the lease refresh, so confirm before acting on it. + if stop.wait(_FIRE_CLAIM_MISS_CONFIRM_SECONDS) or heartbeat_fire_claim( + job_id, expected_owner=owner): + last_confirmed = time.monotonic() + continue lost_ownership.set() logger.warning( "Job '%s': fire claim ownership lost; interrupting stale run", @@ -2645,20 +2653,25 @@ def side_effect_fence(self): return fire_claim_fence(self.job["id"], expected_owner=self.owner) def lost(self) -> bool: - if self.cancel_event is not None and self.cancel_event.is_set(): + if self.transport_cancelled(): return True if self.owner is None: return False if self_removal_delivery_allowed(self.job["id"]): # The run deleted its own record; there is no claim left to re-resolve. return False + # The heartbeat's latched miss is one sample, not the verdict (#113357): the store is + # re-checked here, and a claim that still validates keeps the run's real outcome. The + # owner-fenced mark_job_run / fire_claim_fence remain the authority for side effects. + latched = self.claim_lost is not None and self.claim_lost.is_set() try: if heartbeat_fire_claim(self.job["id"], expected_owner=self.owner): return False except Exception: logger.debug( "Job '%s': fire_claim ownership validation failed", self.job["id"], exc_info=True) - return False + # Unreachable store after a latched miss stays fail-closed; a first miss is best-effort. + return latched if self.claim_lost is not None: self.claim_lost.set() return True diff --git a/tests/cron/test_fire_claim_lost_after_delivery.py b/tests/cron/test_fire_claim_lost_after_delivery.py index 6fe3a145bf4df..9d913dbbf8f69 100644 --- a/tests/cron/test_fire_claim_lost_after_delivery.py +++ b/tests/cron/test_fire_claim_lost_after_delivery.py @@ -1,5 +1,5 @@ -"""#105861: a fire-claim that only *reads* as lost AFTER a run delivered must not overwrite -the delivered run's terminal status — an ok, or a failure notice carrying its real error. +"""#105861 / #113357: a fire-claim that only *reads* as lost must not overwrite a completed run's +terminal status — an ok, or a failure notice carrying its real error. `_FireOwnership.lost()` samples the claim from the store once. When that sample misses after the notice already reached the channel, the run must fall through to the owner-fenced terminal @@ -10,6 +10,7 @@ """ import threading +import time import pytest @@ -159,3 +160,92 @@ def deliver_then_cancel(job, content, **kwargs): record = get_job(job["id"]) assert record["last_status"] == "error" assert record["last_error"] == sched._OWNERSHIP_LOST_INTERRUPTED + + +class _HeartbeatThreadMisses: + """The real ``heartbeat_fire_claim``; the first ``misses`` samples taken on the heartbeat + thread return False (optionally re-owning the stored claim so the loss is genuine).""" + + def __init__(self, real, misses: int, *, steal: bool = False): + self._real, self._left, self._steal = real, misses, steal + self.missed = 0 + + def __call__(self, job_id, *, expected_owner): + if threading.current_thread().name == "cron-fire-claim-heartbeat" and self._left: + self._left -= 1 + self.missed += 1 + if self._steal: + from cron.jobs import _with_job, save_jobs + + def re_own(jobs, _i, job): + job["fire_claim"] = {**job["fire_claim"], "by": "replacement:deadbeef"} + save_jobs(jobs) + + _with_job(job_id, re_own) + return False + return self._real(job_id, expected_owner=expected_owner) + + +def _drive_heartbeat_thread(monkeypatch, *, misses, steal=False): + """run_one_job with the REAL fire-claim heartbeat thread sampling every 10 ms; ``run_job`` + finishes with a complete response once the heartbeat has taken its armed samples.""" + import cron.scheduler as sched + + job = _claimed_job() + hb = _HeartbeatThreadMisses(sched.heartbeat_fire_claim, misses, steal=steal) + delivered, run_cancel = [], [] + + def fake_run_job(job, **kwargs): + deadline = time.monotonic() + 5 + while hb.missed < misses and time.monotonic() < deadline: + time.sleep(0.01) + time.sleep(0.1) # let the heartbeat's confirm sample land + run_cancel.append(kwargs["cancel_event"].is_set()) + return True, "output text", "the report", None + + monkeypatch.setattr(sched, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.01) + monkeypatch.setattr(sched, "_FIRE_CLAIM_MISS_CONFIRM_SECONDS", 0.01, raising=False) + monkeypatch.setattr(sched, "heartbeat_fire_claim", hb) + monkeypatch.setattr(sched, "run_job", fake_run_job) + monkeypatch.setattr(sched, "_deliver_result", lambda job, content, **kw: delivered.append(content)) + return sched, job, hb, delivered, run_cancel + + +@pytest.mark.parametrize("misses, latched", [(1, False), (2, True)], ids=["one-sample", "latched"]) +def test_heartbeat_miss_mid_run_keeps_completed_run(temp_home, monkeypatch, misses, latched): + """#113357: the heartbeat thread's miss is a sample, not the verdict. One miss is re-sampled + and never cancels the run; two consecutive misses latch, but a claim the store still validates + at completion records the completed run — not ``_OWNERSHIP_LOST_INTERRUPTED``.""" + from cron.executions import get_execution + from cron.jobs import get_job + + sched, job, hb, delivered, run_cancel = _drive_heartbeat_thread(monkeypatch, misses=misses) + + assert sched.run_one_job(job) is True + + assert hb.missed == misses + assert run_cancel == [latched], "only a confirmed miss reaches the run's cancel event" + assert delivered == ["the report"] + record = get_job(job["id"]) + assert record["last_status"] == "ok" + assert record["last_error"] is None + assert get_execution(job["execution_id"])["status"] == "completed" + + +def test_confirmed_claim_loss_mid_run_still_yields(temp_home, monkeypatch): + """A genuinely re-owned claim (two misses, stored ``by`` rewritten) still fences the run out: + nothing delivered, no terminal write over the new owner, ledger records the discard.""" + from cron.executions import get_execution + from cron.jobs import get_job + + sched, job, hb, delivered, run_cancel = _drive_heartbeat_thread( + monkeypatch, misses=2, steal=True) + + assert sched.run_one_job(job) is True + + assert run_cancel == [True] + assert delivered == [] + record = get_job(job["id"]) + assert record["last_status"] is None + assert record["fire_claim"]["by"] == "replacement:deadbeef" + assert "discarded" in get_execution(job["execution_id"])["error"] diff --git a/website/docs/developer-guide/cron-internals.md b/website/docs/developer-guide/cron-internals.md index ba7aaa9574317..110dc00d9b2ab 100644 --- a/website/docs/developer-guide/cron-internals.md +++ b/website/docs/developer-guide/cron-internals.md @@ -162,6 +162,17 @@ The same store fields drive every topology: a standalone `hermes -p X gateway run` and a profile served by the default multiplexer (`_start_multiplex` ticks each home under `_profile_cron_scope`) evaluate the identical record. +**Fire-claim lease during a run.** A firing run holds `fire_claim = {at, by}` and +a heartbeat thread refreshes `at` every 60 s (the lease is 300 s). A heartbeat +sample that reads the claim as someone else's is re-sampled once before it +counts: only a confirmed loss cancels the in-flight run. Even then the run's +outcome is decided against the store at completion, not against that latch — +a claim the store still validates records the run's real result (`ok`, or the +real error), while a genuinely re-owned claim discards the stale result and +never writes over the new owner. `Interrupted by shutdown before terminal +completion.` is therefore recorded only when a real transport cancel (gateway +drain) stops a run that still holds its claim. + ### Gateway Integration In gateway mode, the cron **trigger** (the part that decides *when* a due job