fix(cron): bound local fire-fence waits

cron/jobs.py

@@ -105,6 +105,7 @@ _jobs_file_lock = threading.RLock()
 _jobs_lock_state = threading.local()
 _fire_fence_locks: Dict[str, threading.RLock] = {}
 _fire_fence_locks_guard = threading.Lock()
+_fire_fence_lock_state = threading.local()
 
 # Upper bound on waiting for the cross-process .jobs.lock flock (#60703).
 # Every cron function in the process funnels through _jobs_lock(), and the
@@ -387,7 +388,23 @@ def _fire_job_lock(job_id: str):
     with _fire_fence_locks_guard:
         local_lock = _fire_fence_locks.setdefault(lock_key, threading.RLock())
 
-    with local_lock:
+    if not local_lock.acquire(timeout=_JOBS_LOCK_TIMEOUT_SECONDS):
+        logger.error("Timed out waiting for local fire fence %s; failing closed", lock_key)
+        yield False
+        return
+
+    held_locks = getattr(_fire_fence_lock_state, "held", None)
+    if held_locks is None:
+        held_locks = {}
+        _fire_fence_lock_state.held = held_locks
+    if lock_key in held_locks:
+        try:
+            yield held_locks[lock_key]
+        finally:
+            local_lock.release()
+        return
+
+    try:
         ensure_dirs()
         lock_name = uuid.uuid5(uuid.NAMESPACE_URL, lock_key).hex
         lock_path = cron_dir / f".fire-{lock_name}.lock"
@@ -421,9 +438,11 @@ def _fire_job_lock(job_id: str):
         except (OSError, IOError) as exc:
             logger.error("Cron fire fence unavailable for %s: %s", job_id, exc)
 
+        held_locks[lock_key] = acquired
         try:
             yield acquired
         finally:
+            held_locks.pop(lock_key, None)
             if lock_fd is not None:
                 try:
                     if acquired and fcntl is not None:
@@ -436,6 +455,8 @@ def _fire_job_lock(job_id: str):
                     pass
                 finally:
                     lock_fd.close()
+    finally:
+        local_lock.release()
 
 
 @contextlib.contextmanager

tests/cron/test_claim_job_for_fire.py

@@ -215,3 +215,51 @@ def test_fire_claim_fence_rejects_stale_owner(temp_home):
 
     with fire_claim_fence(job["id"], expected_owner="stale") as owns_claim:
         assert owns_claim is False
+
+
+def test_same_process_fire_fence_refuses_second_claim_after_timeout(temp_home, monkeypatch):
+    """A wedged local holder must not indefinitely block another claimant."""
+    import cron.jobs as jobs
+
+    job = jobs.create_job(prompt="x", schedule="every 5m", name="local-fence-timeout")
+    monkeypatch.setattr(jobs, "_JOBS_LOCK_TIMEOUT_SECONDS", 0.1)
+    completed = threading.Event()
+    result = {}
+
+    def second_claimant():
+        result["claimed"] = jobs.claim_job_for_fire(job["id"])
+        completed.set()
+
+    with jobs._fire_job_lock(job["id"]) as acquired:
+        assert acquired is True
+        thread = threading.Thread(target=second_claimant)
+        thread.start()
+        assert completed.wait(timeout=2), "same-process claimant waited past the fire-fence timeout"
+        assert result["claimed"] is False
+
+    thread.join(timeout=2)
+    assert thread.is_alive() is False
+    assert jobs.claim_job_for_fire(job["id"]) is True
+
+
+def test_same_thread_fire_fence_reentrancy_preserves_ownership(temp_home):
+    """Nested same-thread callers retain the existing fire fence."""
+    import cron.jobs as jobs
+
+    job = jobs.create_job(prompt="x", schedule="every 5m", name="local-fence-reentrant")
+    completed = threading.Event()
+    result = {}
+
+    def reentrant_claimant():
+        with jobs._fire_job_lock(job["id"]) as outer_acquired:
+            result["outer"] = outer_acquired
+            with jobs._fire_job_lock(job["id"]) as inner_acquired:
+                result["inner"] = inner_acquired
+        completed.set()
+
+    thread = threading.Thread(target=reentrant_claimant, daemon=True)
+    thread.start()
+    assert completed.wait(timeout=2), "same-thread nested fire fence did not return"
+    assert result == {"outer": True, "inner": True}
+    thread.join(timeout=2)
+    assert thread.is_alive() is False