From 2b1d5900101dfed628445a20b715f8be609eb183 Mon Sep 17 00:00:00 2001 From: Godopu Date: Mon, 14 Sep 2026 00:25:20 +0900 Subject: [PATCH] fix(issue-18): serialize the heartbeat timer lifecycle per job Review round 4 (codex R3): two concurrent `start_timer` calls for the same attempt both passed the "no timer" check, both spawned, and the pidfile tracked only one - the untracked timer could still publish a retained `running` after `completed`. A per-job flock (`.heartbeat.lock`, separate from the registry/seq lock so in-flight beats can still take their seq) now makes observe/drain/spawn/publish-pid one ownership operation; the terminal stop takes the same lock; a start that finds the attempt ended (terminal record or raised barrier) does not arm; the timer removes its own pidfile only under the lock so a replacement's pid is never deleted. Regressions: concurrent starts leave exactly one tracked owner that the terminal stop drains; a start after the barrier arms nothing. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff --- .agents/skills/lib_py/delegation/heartbeat.py | 62 +++++++++++++++-- tests/test_issue18_liveness_contract.py | 66 +++++++++++++++++++ 2 files changed, 124 insertions(+), 4 deletions(-) diff --git a/.agents/skills/lib_py/delegation/heartbeat.py b/.agents/skills/lib_py/delegation/heartbeat.py index c371cb3..b4be8ae 100644 --- a/.agents/skills/lib_py/delegation/heartbeat.py +++ b/.agents/skills/lib_py/delegation/heartbeat.py @@ -41,6 +41,8 @@ checks read. from __future__ import annotations import argparse +import contextlib +import fcntl import json import logging import os @@ -177,6 +179,32 @@ def drain_seconds() -> float: return DEFAULT_DRAIN_S +def lifecycle_lock_path(job_id: str, registry_dir: str) -> str: + return os.path.join(registry_dir, "{}.heartbeat.lock".format(job_id)) + + +@contextlib.contextmanager +def lifecycle_lock(job_id: str, registry_dir: str): + """The per-job timer ownership lock (review round 4, codex R3). + + Observing/draining the previous owner, spawning the replacement and + publishing its pid through the pidfile is one ownership operation, and the + terminal stop takes part in it, so two concurrent `started --heartbeat` + calls cannot both arm a timer with only one of them tracked, and a start + that loses the race against a terminal publish finds the barrier and does + not arm at all. This is a separate flock from the registry/seq lock on + purpose: an in-flight beat needs that lock to take its seq, and the drain + waits for exactly those beats, so holding the seq lock across the drain + would deadlock.""" + os.makedirs(registry_dir, exist_ok=True) + with open(lifecycle_lock_path(job_id, registry_dir), "a+") as fh: + fcntl.flock(fh.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(fh.fileno(), fcntl.LOCK_UN) + + def stop_timer(job_id: str, registry_dir: str, drain: Optional[float] = None) -> bool: """Terminate the job's timer if one runs. Returns True when one was signalled. @@ -184,7 +212,13 @@ def stop_timer(job_id: str, registry_dir: str, drain: Optional[float] = None) -> scheduling and joins its in-flight beats - and SIGKILL it if it has not gone by then, so none of its beats can still publish afterwards. A terminal publish uses this (review round 2, codex R3): "TERM sent" is not "timer - stopped".""" + stopped". Serialised with `start_timer` through the lifecycle lock.""" + with lifecycle_lock(job_id, registry_dir): + return _stop_timer_locked(job_id, registry_dir, drain) + + +def _stop_timer_locked(job_id: str, registry_dir: str, drain: Optional[float] = None) -> bool: + """`stop_timer` body; the caller holds the lifecycle lock.""" pid = running_timer(job_id, registry_dir) path = pidfile_path(job_id, registry_dir) if pid is None or not _is_timer_process(pid, job_id): @@ -264,10 +298,26 @@ def start_timer(job_id: str, registry_dir: str, interval: int, publish that can find only the new timer through the pidfile. ``watch_pid`` binds the timer to a process (default: the nearest agent-CLI ancestor of this process, if any); 0 disables the watch explicitly. + Returns 0 without arming when the attempt already ended (terminal record + or raised barrier): a start that lost the race against the terminal publish + must not leave a timer behind (review round 4, codex R3). """ - stop_timer(job_id, registry_dir, drain=drain_seconds()) if watch_pid is None: watch_pid = nearest_agent_cli_ancestor() + with lifecycle_lock(job_id, registry_dir): + _stop_timer_locked(job_id, registry_dir, drain=drain_seconds()) + try: + status = load_job(job_id, registry_dir).get("status") + except FileNotFoundError: + status = None + if status in TERMINAL_STATUSES or barrier_blocks(job_id, registry_dir, dispatch_token, attempt_id): + logger.info("not arming a heartbeat timer for %s: the attempt already ended", job_id) + return 0 + return _spawn_timer_locked(job_id, registry_dir, interval, dispatch_token, attempt_id, python, watch_pid) + + +def _spawn_timer_locked(job_id, registry_dir, interval, dispatch_token, attempt_id, python, watch_pid) -> int: + """Spawn the timer and publish its pid; the caller holds the lifecycle lock.""" cmd: List[str] = [python or sys.executable, "-m", "lib_py.delegation.heartbeat", "--job", job_id, "--registry-dir", registry_dir, "--interval", str(int(interval))] @@ -388,9 +438,13 @@ def run(job_id: str, registry_dir: str, interval: int, dispatch_token=None, in_flight.append(t) for t in in_flight: t.join(timeout=15) + # Under the lifecycle lock: a replacement may be writing its own pid to + # this file right now, and that owner's pidfile must not be deleted. try: - with open(own_pidfile) as fh: - if fh.read().strip() == str(os.getpid()): + with lifecycle_lock(job_id, registry_dir): + with open(own_pidfile) as fh: + mine = fh.read().strip() == str(os.getpid()) + if mine: os.unlink(own_pidfile) except (OSError, ValueError): pass diff --git a/tests/test_issue18_liveness_contract.py b/tests/test_issue18_liveness_contract.py index 62d8e26..a0f68b4 100644 --- a/tests/test_issue18_liveness_contract.py +++ b/tests/test_issue18_liveness_contract.py @@ -258,6 +258,72 @@ def test_heartbeat_stops_when_the_watched_agent_process_dies(tmp_path): _kill_timer(reg, jid) +def test_concurrent_start_timer_calls_leave_exactly_one_tracked_owner(tmp_path, monkeypatch): + """Review round 4 (codex R3): two `start_timer` calls for the same + job/attempt at once - both see no timer, both spawn, one pidfile - left an + untracked timer that could publish after the terminal event. The lifecycle + lock serialises observe/drain/spawn/publish-pid, so exactly one timer + survives, the pidfile names it, and the terminal stop drains it.""" + from lib_py.delegation import heartbeat, registry + + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent")) + monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs")) + monkeypatch.setenv("MQTT_BROKER", "127.0.0.1") + monkeypatch.setenv("MQTT_PORT", "9") + reg = tmp_path / "jobs" + jid = registry.register_job("race", registry_dir=str(reg)) + tok = _job(reg, jid)["dispatch_token"] + pids = [] + errors = [] + + def start(): + try: + pids.append(heartbeat.start_timer(jid, str(reg), 1, dispatch_token=tok, attempt_id=1, watch_pid=0)) + except Exception as exc: # pragma: no cover - surfaced below + errors.append(exc) + + threads = [threading.Thread(target=start) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=60) + try: + assert not errors, errors + assert len(pids) == 2 and all(pids) + time.sleep(1.0) + alive = [p for p in pids if _alive(p)] + assert len(alive) == 1, f"exactly one timer may survive two concurrent starts, alive={alive} pids={pids}" + assert _timer_pid(reg, jid) == alive[0], "the pidfile names the surviving owner" + # the terminal path drains the tracked owner and nothing else is left + assert heartbeat.stop_timer(jid, str(reg), drain=5.0) is True + assert not any(_alive(p) for p in pids) + assert _timer_pid(reg, jid) is None + finally: + for p in pids: + try: + os.kill(p, signal.SIGKILL) + except OSError: + pass + + +def test_start_timer_after_the_terminal_barrier_does_not_arm(tmp_path, monkeypatch): + """A `started --heartbeat` that loses the race against the terminal + publish must not leave a timer behind.""" + from lib_py.delegation import heartbeat, registry + + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent")) + monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs")) + reg = tmp_path / "jobs" + jid = registry.register_job("late-start", registry_dir=str(reg)) + tok = _job(reg, jid)["dispatch_token"] + heartbeat.raise_barrier(jid, str(reg), dispatch_token=tok, attempt_id=1) + assert heartbeat.start_timer(jid, str(reg), 1, dispatch_token=tok, attempt_id=1, watch_pid=0) == 0 + assert _timer_pid(reg, jid) is None + assert not (reg / f"{jid}.heartbeat.pid").exists() + + def test_stop_timer_never_signals_a_recycled_pid(tmp_path, monkeypatch): """A stale pidfile (timer crashed) may name a pid that now belongs to something else; stop_timer must only signal a heartbeat timer of this job."""