fix(cron): a completed run keeps its result when a fire-claim heartbeat sample misses (#113357, #105861, salvage #113364)

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

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"]

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