fix(update): the crash cells' marker oracle applies update_lock's v1 age ceiling (review F51)

What: the test's _identity_live takes the marker age and, for an identity without a creation time, is live only within hermes_cli.update_lock.UPDATE_MARKER_MAX_AGE_SECONDS.
Why: the oracle read a ct-less (v1) marker for a live pid as LIVE past the product's 20-minute ceiling, so it could disagree with the verdict update_lock gives every other updater.
Red: pre-fix test_crash_cells.py -> [1260-False-False] FAILED (6 passed, 1 failed).
Green: 7 passed, 0 failed.
This commit is contained in:
Teknium
2026-10-05 00:53:27 -07:00
parent 4da5a08c84
commit 46efbcbe28
2 changed files with 24 additions and 4 deletions

View File

@@ -1,9 +1,12 @@
"""The Windows crash cells' oracles, judged on this OS: a verdict the cells can only reach on a
Windows runner must still say pass exactly when the user's update was what the cell claims."""
import os
import subprocess
import time
from pathlib import Path
import psutil
import pytest
import tests.e2e.core.windows_update.test_crash_cells as crash
@@ -57,3 +60,15 @@ def test_kill_point_counts_only_an_update_killed_while_running(
else:
with pytest.raises(AssertionError, match="cell: the update"):
crash._kill_when(_Machine(tmp_path), proc, "cell", point, "t")
@pytest.mark.parametrize("age,with_ct,live", [
(60, False, True), # v1 marker inside update_lock's 20-minute ceiling
(1260, False, False), # v1 past the ceiling: the product reads it DEAD (the pid may be reused)
(1260, True, True), # a matching creation time is live at any age
])
def test_marker_oracle_reads_a_marker_live_exactly_when_update_lock_does(age, with_ct, live):
pid = os.getpid()
ct = f"ct:{psutil.Process(pid).create_time():.3f}\n" if with_ct else ""
text = f"{pid}\n{time.time() - age}\n{ct}"
assert crash._marker_live(text) == (f"owner {pid}" if live else None)

View File

@@ -67,6 +67,7 @@ from pathlib import Path
import psutil
import pytest
from hermes_cli.update_lock import UPDATE_MARKER_MAX_AGE_SECONDS
from tests.e2e.core._pending_fixes import known_failure
from tests.e2e.core.windows._helpers import _decode
@@ -404,14 +405,17 @@ def _parse_marker(text: str) -> dict:
"delegate": delegate, "delegate_ct": delegate_ct}
def _identity_live(pid: int | None, ct: float | None) -> bool:
def _identity_live(pid: int | None, ct: float | None, age: float) -> bool:
"""update_lock's C1 verdict for one identity of a marker ``age`` seconds old."""
if not pid or pid <= 0:
return False
try:
proc = psutil.Process(pid)
if not proc.is_running():
return False
return ct is None or abs(proc.create_time() - ct) <= CT_TOLERANCE
if ct is None: # v1 / no creation time: the pid may be reused, so only the age ceiling bounds it
return age <= UPDATE_MARKER_MAX_AGE_SECONDS
return abs(proc.create_time() - ct) <= CT_TOLERANCE
except psutil.Error:
return False
@@ -419,9 +423,10 @@ def _identity_live(pid: int | None, ct: float | None) -> bool:
def _marker_live(text: str) -> str | None:
"""Who keeps the marker LIVE (``"owner <pid>"`` / ``"delegate <pid>"``), or None."""
m = _parse_marker(text)
if _identity_live(m["pid"], m["ct"]):
age = time.time() - (m["started_at"] or 0.0)
if _identity_live(m["pid"], m["ct"], age):
return f"owner {m['pid']}"
if _identity_live(m["delegate"], m["delegate_ct"]):
if _identity_live(m["delegate"], m["delegate_ct"], age):
return f"delegate {m['delegate']}"
return None