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)