f00b4a0339
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.
255 lines
12 KiB
Python
255 lines
12 KiB
Python
"""Real-process pause/consume/recovery contracts, not native Windows SCM evidence."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
import time
|
|
|
|
import pytest
|
|
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def _wait(path, process):
|
|
deadline = time.monotonic() + 30
|
|
while not path.exists():
|
|
assert process.poll() is None, process.communicate()
|
|
assert time.monotonic() < deadline, str(path)
|
|
time.sleep(0.01)
|
|
|
|
|
|
@pytest.fixture
|
|
def cell(tmp_path):
|
|
# An unchanged copy owns its private checkout lock/root; never the live install's.
|
|
root = tmp_path / "checkout"
|
|
(root / "hermes_cli").mkdir(parents=True)
|
|
shutil.copyfile(REPO / "hermes_cli/update_pause_record.py", root / "hermes_cli/update_pause_record.py")
|
|
home = tmp_path / "home"
|
|
(home / ".hermes").mkdir(parents=True)
|
|
env = {**os.environ, "HOME": str(home), "USERPROFILE": str(home),
|
|
"HERMES_HOME": str(home / ".hermes"), "PYTHONPATH": str(REPO),
|
|
"PYTHONDONTWRITEBYTECODE": "1"}
|
|
children = []
|
|
|
|
def launch(code, *args):
|
|
script = tmp_path / f"child-{len(children)}.py"
|
|
script.write_text(textwrap.dedent('''
|
|
import importlib.util, json, os, sys, time
|
|
from pathlib import Path
|
|
root = Path(sys.argv[1])
|
|
spec = importlib.util.spec_from_file_location("hermes_cli.update_pause_record", root / "hermes_cli/update_pause_record.py")
|
|
r = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = r
|
|
spec.loader.exec_module(r)
|
|
''') + textwrap.dedent(code), encoding="utf-8")
|
|
proc = subprocess.Popen([sys.executable, str(script), str(root), *map(str, args)],
|
|
cwd=tmp_path, env=env, stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
|
children.append(proc)
|
|
return proc
|
|
|
|
yield tmp_path, home / ".hermes", launch
|
|
for proc in children:
|
|
if proc.poll() is None:
|
|
proc.kill()
|
|
proc.communicate(timeout=10)
|
|
|
|
|
|
def _result(proc):
|
|
out, err = proc.communicate(timeout=30)
|
|
assert proc.returncode == 0, (out, err)
|
|
return json.loads(out.splitlines()[-1])
|
|
|
|
|
|
@pytest.mark.parametrize("claim_first", [False, True])
|
|
def test_consumption_checkpoints_debt_before_unlink_and_producer_death(cell, claim_first):
|
|
base, home, launch = cell
|
|
consumer = launch('''
|
|
from gateway.status import consume_planned_stop_marker_for_self
|
|
Path(sys.argv[2]).write_text(str(os.getpid()), encoding="utf-8")
|
|
while not Path(sys.argv[3]).exists(): time.sleep(.01)
|
|
accepted = consume_planned_stop_marker_for_self()
|
|
Path(sys.argv[4]).write_text(json.dumps(accepted), encoding="utf-8")
|
|
while True: time.sleep(.05)
|
|
''', base / "ready", base / "consume", base / "consumed")
|
|
_wait(base / "ready", consumer)
|
|
owner = launch('''
|
|
from hermes_cli import update_cmd_windows as w
|
|
from types import SimpleNamespace
|
|
pid = int(sys.argv[2])
|
|
home = Path(os.environ["HERMES_HOME"])
|
|
token = r.record_pause({"resume_needed": True, "profiles": {"default": pid},
|
|
"identities": {str(pid): r.identity(pid)["ct"]}}, None, [])
|
|
# Discovery and the gateway can spell the same profile path differently.
|
|
r.mark_stop_requested(token, [pid], {pid: home / ".." / ".hermes" / ".gateway-planned-stop.json"})
|
|
# The real callback is the first operation after marker publication. Hold it
|
|
# before its checkpoint while another interpreter runs the actual consumer.
|
|
def on_request(pid):
|
|
Path(sys.argv[3]).touch()
|
|
while True: time.sleep(.05)
|
|
w._request_socket_pauses([pid], {pid: SimpleNamespace(profile="default", path=home)},
|
|
set(), on_request=on_request)
|
|
''', consumer.pid, base / "published")
|
|
_wait(base / "published", owner)
|
|
recovery = None
|
|
if claim_first:
|
|
owner.kill()
|
|
owner.communicate(timeout=10)
|
|
recovery = launch('''
|
|
won = r.claim(r.record_path())
|
|
assert won is not None
|
|
Path(sys.argv[2]).touch()
|
|
while not Path(sys.argv[3]).exists(): time.sleep(.01)
|
|
r._resume_claimed(*won)
|
|
print(json.dumps([body["token"] for _, body in r.orphans()]))
|
|
''', base / "claimed", base / "recover")
|
|
_wait(base / "claimed", recovery)
|
|
(base / "consume").touch()
|
|
_wait(base / "consumed", consumer)
|
|
assert json.loads((base / "consumed").read_text(encoding="utf-8-sig")) is True
|
|
assert not (home / ".gateway-planned-stop.json").exists()
|
|
if owner.poll() is None:
|
|
owner.kill()
|
|
owner.communicate(timeout=10)
|
|
assert consumer.poll() is None
|
|
if recovery is not None:
|
|
(base / "recover").touch()
|
|
else:
|
|
recovery = launch('''
|
|
r.recover(["status"])
|
|
print(json.dumps([body["token"] for _, body in r.orphans()]))
|
|
''')
|
|
after = _result(recovery)
|
|
assert len(after) == 1, "consumed request lost all durable restart debt"
|
|
assert after[0]["profiles"] == {"default": consumer.pid}
|
|
assert after[0]["stop_sent"] == [str(consumer.pid)]
|
|
|
|
|
|
def test_failed_consumer_checkpoint_keeps_request_and_user_stop_is_not_update_debt(cell):
|
|
_, _, launch = cell
|
|
result = _result(launch('''
|
|
from gateway import status
|
|
from hermes_cli.update_cmd_windows import _write_update_planned_stop_marker
|
|
pid = os.getpid()
|
|
home = Path(os.environ["HERMES_HOME"])
|
|
path = home / ".gateway-planned-stop.json"
|
|
token = r.record_pause({"resume_needed": True, "profiles": {"default": pid},
|
|
"identities": {str(pid): r.identity(pid)["ct"]}}, None, [])
|
|
r.mark_stop_requested(token, [pid], {pid: path})
|
|
assert _write_update_planned_stop_marker(home, pid)
|
|
# An actual directory at the atomic writer's temp path rejects publication.
|
|
obstruction = r.record_path().with_name(f"{r.record_path().name}.{pid}.tmp")
|
|
obstruction.mkdir()
|
|
# The checkpoint cannot land: the stop is still planned, and the request stays on disk.
|
|
try:
|
|
planned = status.consume_planned_stop_marker_for_self()
|
|
except OSError as exc:
|
|
planned = type(exc).__name__
|
|
retained = path.exists() and r.read()["token"]["stop_sent"] == []
|
|
obstruction.rmdir()
|
|
marker = json.loads(path.read_text(encoding="utf-8-sig"))
|
|
path.write_text(json.dumps({**marker, "stopper_pid": pid + 1}), encoding="utf-8")
|
|
user_consumed = status.consume_planned_stop_marker_for_self()
|
|
user_debt = r.read()["token"]["stop_sent"]
|
|
assert _write_update_planned_stop_marker(home, pid)
|
|
accepted = status.consume_planned_stop_marker_for_self()
|
|
print(json.dumps({"planned": planned, "retained": retained, "user_consumed": user_consumed,
|
|
"user_debt": user_debt, "accepted": accepted, "sent": r.read()["token"]["stop_sent"],
|
|
"pid": pid}))
|
|
'''))
|
|
assert result == {"planned": True, "retained": True, "user_consumed": True, "user_debt": [],
|
|
"accepted": True, "sent": [str(result["pid"])], "pid": result["pid"]}
|
|
|
|
|
|
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)
|
|
''', base / "held", base / "release")
|
|
_wait(base / "held", holder)
|
|
try:
|
|
result = _result(launch('''
|
|
from gateway import status
|
|
from hermes_cli.update_cmd_windows import _write_update_planned_stop_marker
|
|
pid = os.getpid()
|
|
home = Path(os.environ["HERMES_HOME"])
|
|
assert _write_update_planned_stop_marker(home, pid)
|
|
started = time.monotonic()
|
|
try:
|
|
planned = status.consume_planned_stop_marker_for_self()
|
|
except OSError as exc:
|
|
planned = type(exc).__name__
|
|
print(json.dumps({"planned": planned, "seconds": time.monotonic() - started,
|
|
"retained": (home / ".gateway-planned-stop.json").exists()}))
|
|
'''))
|
|
finally:
|
|
(base / "release").touch()
|
|
assert result["planned"] is True and result["retained"], result
|
|
assert result["seconds"] < 5, f"the gateway's shutdown blocked {result['seconds']:.1f}s on the pause mutex"
|
|
|
|
|
|
def test_recovery_only_accepts_markers_the_live_consumer_accepts(cell):
|
|
_, _, launch = cell
|
|
result = _result(launch('''
|
|
from datetime import datetime, timedelta, timezone
|
|
from gateway import status
|
|
pid = os.getpid()
|
|
home = Path(os.environ["HERMES_HOME"])
|
|
marker_path = home / ".gateway-planned-stop.json"
|
|
token = {"profiles": {"default": pid}, "identities": {str(pid): r.identity(pid)["ct"]},
|
|
"stopper_pid": pid, "stop_requested": [str(pid)], "stop_markers": {str(pid): str(marker_path)}}
|
|
current = status.get_process_start_time(pid)
|
|
assert current is not None
|
|
fresh = {"target_pid": pid, "stopper_pid": pid, "target_start_time": current,
|
|
"written_at": datetime.now(timezone.utc).isoformat()}
|
|
cases = {"valid": fresh,
|
|
"expired": {**fresh, "written_at": (datetime.now(timezone.utc) - timedelta(seconds=120)).isoformat()},
|
|
"wrong_incarnation": {**fresh, "target_start_time": current + 1},
|
|
"malformed": {"target_pid": pid, "stopper_pid": pid}}
|
|
out = {}
|
|
for name, marker in cases.items():
|
|
marker_path.write_text(json.dumps(marker), encoding="utf-8")
|
|
kept = r.drop_never_stopped(dict(token))
|
|
consumed = status.consume_planned_stop_marker_for_self()
|
|
out[name] = {"kept": bool(kept["profiles"]), "consumed": consumed}
|
|
print(json.dumps(out))
|
|
'''))
|
|
assert result["valid"] == {"kept": True, "consumed": True}
|
|
for name in ("expired", "wrong_incarnation", "malformed"):
|
|
assert result[name] == {"kept": False, "consumed": False}, (name, result)
|
|
|
|
|
|
def test_orphan_adoption_preserves_cold_start_generation_and_current_precedence(cell):
|
|
_, _, launch = cell
|
|
_result(launch('''
|
|
token = {"resume_needed": True, "profiles": {}, "cold_start_if_installed": True,
|
|
"attested_generation": "old-active", "cold_start_profiles": {"beta": "old-beta", "gamma": "old-gamma"}}
|
|
r.write(r.stamp_tree(token), owner=r.UNOWNED)
|
|
print(json.dumps(r.read()["token"]))
|
|
'''))
|
|
adopted = _result(launch('''
|
|
adopted, claims = r.adopt_orphans()
|
|
token = {"resume_needed": True, "profiles": {}, "cold_start_profiles": {"beta": "current-beta"}}
|
|
r.record_pause(token, adopted, claims)
|
|
print(json.dumps(r.read()["token"]))
|
|
'''))
|
|
assert adopted.get("cold_start_if_installed") is True
|
|
assert adopted.get("attested_generation") == "old-active"
|
|
assert adopted.get("cold_start_profiles") == {"beta": "current-beta", "gamma": "old-gamma"}
|
|
again = _result(launch('''
|
|
adopted, claims = r.adopt_orphans()
|
|
r.record_pause({"resume_needed": True, "profiles": {}, "cold_start_if_installed": True,
|
|
"attested_generation": "current-active"}, adopted, claims)
|
|
print(json.dumps(r.read()["token"]))
|
|
'''))
|
|
assert again["attested_generation"] == "current-active"
|
|
assert again["cold_start_profiles"] == adopted["cold_start_profiles"]
|