From f00b4a0339a42224ef57c30492fc04437bd1dbf7 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Mon, 5 Oct 2026 23:09:16 -0700 Subject: [PATCH] fix: the planned-stop consumer takes the pause lock only for a pause on disk and drops settled receipts consume_planned_stop_marker_for_self took .hermes-update-paused-gateways.lock on every platform for every stop (up to 2 s when busy), creating the lock file in the root home, and on a busy lock fell back to keeping the request and writing a .accepted receipt that nothing ever removed. The lock only protects the checkpoint into a Windows update pause record; with no record or claim of this checkout on disk (every non-Windows host, or a pause recovery/the updater already retired) there is nothing to checkpoint. The consumer now holds the lock only when a pause is on disk. The checkpoint hook re-checks after the request was read, so a pause that appeared after the first look (an update writes its record before its request) is still checkpointed under the lock. With no pause on disk, the consumer removes the .accepted receipt and an expired stamped request an earlier drain left: their debt is settled. While any pause is on disk both stay (never earlier). test_a_busy_pause_mutex_never_turns_a_planned_stop_into_an_unplanned_one now puts a pause record on disk first: without one, keeping the request on a busy lock was the leftover this fixes; with one, its assertions (planned, request retained, no block) are unchanged. --- gateway/status.py | 44 +++++++++++++---- .../test_planned_stop_accepted_debt.py | 47 +++++++++++++++++++ .../test_update_pause_boundaries.py | 1 + 3 files changed, 84 insertions(+), 8 deletions(-) diff --git a/gateway/status.py b/gateway/status.py index d0ec62ab88..3275cf5cfa 100644 --- a/gateway/status.py +++ b/gateway/status.py @@ -1682,7 +1682,7 @@ def _read_live_pid_marker(path: Path, ttl_s: int) -> Optional[tuple[dict[str, An return None target_pid = _pid_from_record(record, "target_pid") if target_pid is None or _marker_is_stale(record.get("written_at") or "", ttl_s): - if record.get("accepted") is not True: + if record.get("accepted") is not True or not _update_pause_on_disk(): _unlink_quietly(path) return None return record, target_pid, record.get("target_start_time") @@ -1961,17 +1961,45 @@ def write_planned_stop_marker(target_pid: int) -> bool: _PLANNED_STOP_MUTEX_WAIT_S = 2.0 +def _update_pause_on_disk() -> bool: + """A Windows update pause of this checkout is on disk (its record or a recovery's claim), or the + record directory cannot be listed. Only then does a consume need the pause lock and a checkpoint, + and only then can accepted-stop evidence beside a request still be owed: with none, every pause + was settled by recovery or the updater (or never made: every non-Windows host).""" + from hermes_cli import update_pause_record + record = update_pause_record.record_path() + try: + with os.scandir(record.parent) as entries: + return any(e.name == record.name or (e.name.startswith(record.name + ".") and e.name.endswith(".claim")) + for e in entries) + except FileNotFoundError: + return False + except OSError: + return True + + +def _checkpoint_planned_stop(path: Path, record: dict[str, Any]) -> None: + """``on_consume``, decided after the request was read: an update writes its record before its + request, so a pause that appeared since the consumer first looked is checkpointed too.""" + from hermes_cli import update_pause_record + if _update_pause_on_disk(): + with update_pause_record._mutex(_PLANNED_STOP_MUTEX_WAIT_S): + update_pause_record.mark_stop_consumed(path, record) + + def consume_planned_stop_marker_for_self() -> bool: """Return True when the current process is being intentionally stopped.""" from hermes_cli import update_pause_record - # Recovery must see either the request or its checkpoint, never the gap - # between validating/consuming the marker and scheduling asynchronous stop. + path = _get_planned_stop_marker_path() try: - with update_pause_record._mutex(_PLANNED_STOP_MUTEX_WAIT_S): - return _consume_pid_marker_for_self( - _get_planned_stop_marker_path(), ttl_s=_PLANNED_STOP_MARKER_TTL_S, - on_consume=update_pause_record.mark_stop_consumed, - ) + paused = _update_pause_on_disk() + if not paused: # an earlier drain's receipt: its debt is settled + _unlink_quietly(update_pause_record._accepted_path(path)) + # With a pause on disk, recovery must see either the request or its checkpoint, never the + # gap between validating/consuming the marker and scheduling asynchronous stop. + with update_pause_record._mutex(_PLANNED_STOP_MUTEX_WAIT_S) if paused else contextlib.nullcontext(): + return _consume_pid_marker_for_self(path, ttl_s=_PLANNED_STOP_MARKER_TTL_S, + on_consume=_checkpoint_planned_stop) except OSError as exc: # The pause bookkeeping (a busy mutex, a checkpoint that cannot be written) never decides # whether this stop was planned: classify without consuming, and leave a receipt beside the diff --git a/tests/gateway/test_planned_stop_accepted_debt.py b/tests/gateway/test_planned_stop_accepted_debt.py index 7315ab811f..aa7672ab59 100644 --- a/tests/gateway/test_planned_stop_accepted_debt.py +++ b/tests/gateway/test_planned_stop_accepted_debt.py @@ -10,6 +10,8 @@ from __future__ import annotations import json import os import sys +import threading +import time from datetime import datetime, timedelta, timezone from pathlib import Path @@ -80,3 +82,48 @@ def test_a_stamped_accepted_request_outlives_its_ttl_while_the_pause_is_owed(tmp assert status.planned_stop_marker_targets_self() is False assert marker.exists(), f"the {second_look} deleted the stamped accepted request" assert _owed(saved["pause_id"]) == {"p": os.getpid()}, "an accepted stop lost its restart debt" + + +@pytest.mark.parametrize("home_writable", [True, False], ids=["receipt", "stamped_request"]) +def test_accepted_stop_evidence_is_removed_once_its_pause_is_settled_and_never_before( + tmp_path, monkeypatch, home_writable): + """``.accepted`` receipts and stamped requests are cleaned up by the next consume once no pause of + this checkout is on disk (recovery or the updater retired it), and kept while one still is.""" + marker, saved = _accepted_unrecorded(tmp_path, monkeypatch, home_writable=home_writable) + evidence = pause_record._accepted_path(marker) if home_writable else marker + status.consume_planned_stop_marker_for_self() + assert evidence.exists(), "accepted-stop evidence removed while its pause is still owed" + pause_record.discharge(saved) + assert not pause_record.record_path().exists(), "premise: the pause is settled" + status.consume_planned_stop_marker_for_self() + assert not pause_record._accepted_path(marker).exists(), "a settled receipt was left behind" + assert not marker.exists(), "a settled stamped request was left behind" + + +def test_with_no_pause_on_disk_the_consumer_never_waits_on_the_pause_lock(tmp_path, monkeypatch): + """No pause record (every non-Windows host, or a settled pause): a plain ``hermes gateway stop`` is + consumed without the pause lock, so another holder costs it no wait and leaves no request or + ``.accepted`` receipt behind.""" + _home(tmp_path, monkeypatch) + held, release = threading.Event(), threading.Event() + + def holder(): + with pause_record._mutex(): + held.set() + release.wait(10) + + thread = threading.Thread(target=holder) + thread.start() + try: + assert held.wait(5), "premise: another thread holds the pause lock" + marker = status._get_planned_stop_marker_path() + assert status.write_planned_stop_marker(os.getpid()) + started = time.monotonic() + assert status.consume_planned_stop_marker_for_self() is True + elapsed = time.monotonic() - started + finally: + release.set() + thread.join() + assert elapsed < 1.0, f"the consumer waited {elapsed:.2f}s on a pause lock no pause needed" + assert not marker.exists(), "the consumed request was left on disk" + assert not pause_record._accepted_path(marker).exists(), "an .accepted receipt was written with no pause" diff --git a/tests/hermes_cli/test_update_pause_boundaries.py b/tests/hermes_cli/test_update_pause_boundaries.py index 0da9e794c3..9dae296ca0 100644 --- a/tests/hermes_cli/test_update_pause_boundaries.py +++ b/tests/hermes_cli/test_update_pause_boundaries.py @@ -169,6 +169,7 @@ def test_failed_consumer_checkpoint_keeps_request_and_user_stop_is_not_update_de def test_a_busy_pause_mutex_never_turns_a_planned_stop_into_an_unplanned_one(cell): base, _, launch = cell holder = launch(''' + r.write({"pause_id": "busy", "resume_needed": True}) # a pause on disk: the consume needs the mutex with r._mutex(): Path(sys.argv[2]).touch() while not Path(sys.argv[3]).exists(): time.sleep(.05)