275 lines
11 KiB
Python
275 lines
11 KiB
Python
"""Lease-failure recovery contracts for the execution-pool coordinator.
|
|
|
|
The independent review reproduced a wedge in which a Kanban write that failed
|
|
during lease expiry left a ``lease_failed`` row that was invisible, immortal,
|
|
and fatal. These tests pin the recovered behaviour: the row stays retryable, is
|
|
only collected on authoritative terminal evidence, and survives a restart on the
|
|
PVC the store lives on.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).parents[2]
|
|
SCRIPTS = ROOT / "services/hermes/scripts"
|
|
SCM_SCRIPTS = ROOT / "services/hermes/scm-common/scripts"
|
|
sys.path[:0] = [str(SCRIPTS), str(SCM_SCRIPTS)]
|
|
|
|
import execution_pool_coordinator as coordinator # noqa: E402
|
|
import execution_pool_maintenance as maintenance # noqa: E402
|
|
import execution_pool_store as pool_store # noqa: E402
|
|
from testing.tests.test_hermes_execution_pool_coordinator_v2 import ( # noqa: E402
|
|
MASTER,
|
|
assignment_payload,
|
|
binding,
|
|
install_kanban,
|
|
task,
|
|
)
|
|
from testing.tests.test_hermes_execution_pool_support import ( # noqa: E402
|
|
LOCKED,
|
|
age,
|
|
clear_pool_deferrals, # noqa: F401 - autouse fixture, resolved by name
|
|
exhausted_store,
|
|
states,
|
|
)
|
|
|
|
|
|
def test_locked_board_at_lease_expiry_stays_retryable_and_never_wedges(
|
|
tmp_path, monkeypatch, capsys
|
|
):
|
|
"""The exact review repro: a locked board must not strand the run."""
|
|
live = task()
|
|
kanban = install_kanban(monkeypatch, [live], tmp_path)
|
|
monkeypatch.setattr(
|
|
coordinator, "assignment_payload",
|
|
lambda _db, _connection, _task, _board: assignment_payload(),
|
|
)
|
|
|
|
def locked(*_args, **_kwargs):
|
|
raise LOCKED
|
|
|
|
kanban.block_task = locked
|
|
store = exhausted_store(tmp_path / "wedge.db")
|
|
pool = coordinator.Coordinator(MASTER, store)
|
|
|
|
pool.expire_leases()
|
|
assert states(store) == [("23", 0, 3, "lease_failed")]
|
|
assert store.available_ordinals() == [0, 1, 2]
|
|
assert "lease park" in capsys.readouterr().err
|
|
|
|
# 1. The row is never collected while its outcome is unconfirmed.
|
|
age(store, 10 * 365 * 86400)
|
|
assert store.garbage_collect(0) == 0
|
|
assert states(store) == [("23", 0, 3, "lease_failed")]
|
|
|
|
# 2. reconcile() no longer hits the conflicting-duplicate primary key.
|
|
for _cycle in range(3):
|
|
pool.reconcile()
|
|
assert states(store) == [("23", 0, 3, "lease_failed")]
|
|
|
|
# 3. dispatch() makes no spurious Kanban mutation.
|
|
monkeypatch.setattr(
|
|
maintenance.cli_lane_dispatch, "claim_ready",
|
|
lambda *_args, **_kwargs: [("metis", live.id)],
|
|
)
|
|
pool.dispatch()
|
|
assert kanban.blocked == []
|
|
assert states(store) == [("23", 0, 3, "lease_failed")]
|
|
|
|
# 4. Once the board recovers, the retry records the exact run and settles.
|
|
kanban.block_task = lambda _connection, task_id, **values: (
|
|
kanban.blocked.append((task_id, values)) or True
|
|
)
|
|
pool.expire_leases()
|
|
assert kanban.blocked[-1][1] == {
|
|
"reason": "Distributed worker lease expired after 3 fenced attempts",
|
|
"kind": "transient",
|
|
"expected_run_id": 23,
|
|
}
|
|
assert states(store) == [("23", 0, 3, "finalized")]
|
|
|
|
# 5. Only now, with authoritative evidence, is the row collectable.
|
|
age(store, 10 * 365 * 86400)
|
|
assert store.garbage_collect(0) == 1
|
|
assert states(store) == []
|
|
|
|
|
|
def test_refused_exact_run_park_keeps_the_row_for_the_next_pass(tmp_path, monkeypatch):
|
|
"""No terminal state without evidence, even when the write is refused."""
|
|
live = task()
|
|
kanban = install_kanban(monkeypatch, [live], tmp_path)
|
|
kanban.block_task = lambda _connection, task_id, **values: (
|
|
kanban.blocked.append((task_id, values)) or False
|
|
)
|
|
store = exhausted_store(tmp_path / "refused.db")
|
|
pool = coordinator.Coordinator(MASTER, store)
|
|
|
|
pool.expire_leases()
|
|
assert states(store) == [("23", 0, 3, "lease_failed")]
|
|
assert len(kanban.blocked) == 1
|
|
|
|
# The next pass retries rather than giving up on the run.
|
|
pool.expire_leases()
|
|
assert len(kanban.blocked) == 2
|
|
assert states(store) == [("23", 0, 3, "lease_failed")]
|
|
|
|
|
|
def test_lease_failure_settles_stale_when_the_run_moved_or_vanished(
|
|
tmp_path, monkeypatch
|
|
):
|
|
moved = task(current_run_id=99)
|
|
kanban = install_kanban(monkeypatch, [moved], tmp_path)
|
|
store = exhausted_store(tmp_path / "moved.db")
|
|
coordinator.Coordinator(MASTER, store).expire_leases()
|
|
assert states(store) == [("23", 0, 3, "stale")]
|
|
assert kanban.blocked == []
|
|
|
|
kanban.tasks.clear()
|
|
gone = exhausted_store(tmp_path / "gone.db", binding(task_id="t_gone", attempt=3))
|
|
coordinator.Coordinator(MASTER, gone).expire_leases()
|
|
assert states(gone) == [("23", 0, 3, "stale")]
|
|
|
|
|
|
def test_lease_failure_settles_finalized_when_kanban_already_parked_the_run(
|
|
tmp_path, monkeypatch
|
|
):
|
|
parked = task(status="blocked")
|
|
kanban = install_kanban(monkeypatch, [parked], tmp_path)
|
|
store = exhausted_store(tmp_path / "parked.db")
|
|
coordinator.Coordinator(MASTER, store).expire_leases()
|
|
assert states(store) == [("23", 0, 3, "finalized")]
|
|
assert kanban.blocked == []
|
|
|
|
|
|
def test_noncanonical_run_lease_failure_is_released_without_a_board_call(
|
|
tmp_path, monkeypatch
|
|
):
|
|
kanban = install_kanban(monkeypatch, [task()], tmp_path)
|
|
store = exhausted_store(
|
|
tmp_path / "bad-run.db", binding(run_id="run-bad", attempt=3)
|
|
)
|
|
coordinator.Coordinator(MASTER, store).expire_leases()
|
|
assert states(store) == [("run-bad", 0, 3, "stale")]
|
|
assert kanban.blocked == []
|
|
|
|
|
|
def test_lease_expiry_releases_only_a_trusted_publication_retry(tmp_path, monkeypatch):
|
|
"""An OOM-style lease expiry returns one receipt before parking the native run."""
|
|
kanban = install_kanban(monkeypatch, [task()], tmp_path)
|
|
store = pool_store.PoolStore(tmp_path / "resume-lease.db")
|
|
exact = binding(attempt=3)
|
|
store.add(exact, assignment_payload(scm_resume={"source": {"worker_ordinal": 0}}))
|
|
store.offer(0)
|
|
with store._connect() as connection:
|
|
connection.execute("UPDATE assignments SET lease_until=1")
|
|
released = []
|
|
monkeypatch.setattr(
|
|
coordinator.supervisor_state, "reissue_publication_retry",
|
|
lambda *values: released.append(values) or True,
|
|
)
|
|
coordinator.Coordinator(MASTER, store).expire_leases()
|
|
assert released and released[0][2] == "23"
|
|
assert kanban.blocked[-1][1]["kind"] == "transient"
|
|
|
|
|
|
def test_a_durable_row_in_any_state_blocks_conflicting_re_adoption(tmp_path, monkeypatch):
|
|
"""known_runs() covers terminal and retrying rows, not just live ones."""
|
|
live = task()
|
|
install_kanban(monkeypatch, [live], tmp_path)
|
|
monkeypatch.setattr(
|
|
coordinator, "assignment_payload",
|
|
lambda _db, _connection, _task, _board: assignment_payload(context="different"),
|
|
)
|
|
store = pool_store.PoolStore(tmp_path / "known.db")
|
|
store.add(binding(), assignment_payload())
|
|
pool = coordinator.Coordinator(MASTER, store)
|
|
for state in ("lease_failed", "finalized", "stale"):
|
|
with store._connect() as connection:
|
|
connection.execute("UPDATE assignments SET state=?", (state,))
|
|
assert store.known_runs() == {("metis", "t_deadbeef", "23")}
|
|
pool.reconcile()
|
|
assert states(store) == [("23", 0, 1, state)]
|
|
|
|
|
|
def test_lease_fence_records_exact_terminal_evidence_idempotently(tmp_path):
|
|
"""An expired publication run becomes schedulable only after fenced evidence."""
|
|
store = pool_store.PoolStore(tmp_path / "lease-evidence.db")
|
|
exact = binding(attempt=3)
|
|
receipt = {
|
|
"source": {"board": "metis", "task_id": "t_deadbeef", "worker_ordinal": 0},
|
|
"result_digest": "a" * 64,
|
|
}
|
|
store.add(exact, assignment_payload(scm_resume=receipt))
|
|
with store._connect() as connection:
|
|
connection.execute("UPDATE assignments SET state='lease_failed'")
|
|
marker = "[hermes-publication-retry-fence:23:" + "a" * 64 + "]"
|
|
assert store.record_publication_lease_failure(exact, receipt, marker)
|
|
assert store.record_publication_lease_failure(exact, receipt, marker)
|
|
terminal = store.terminal_record("metis", "t_deadbeef", "23")
|
|
assert terminal is None # Evidence alone never releases the terminal state.
|
|
assert store.finalize(exact, "finalized")
|
|
result = store.terminal_record("metis", "t_deadbeef", "23")["result"]
|
|
assert result["capacity_failure"] is True and result["publication_retry_lease_fence"] == marker
|
|
|
|
|
|
def test_lease_expiry_auto_retry_evidence_requires_fenced_native_outcome(tmp_path, monkeypatch):
|
|
"""A fenced OOM park gains exact terminal evidence for the one retry scheduler."""
|
|
kanban = install_kanban(monkeypatch, [task()], tmp_path)
|
|
store = pool_store.PoolStore(tmp_path / "lease-auto-retry.db")
|
|
exact = binding(attempt=3)
|
|
receipt = {
|
|
"source": {"board": "metis", "task_id": "t_deadbeef", "worker_ordinal": 0},
|
|
"result_digest": "d" * 64,
|
|
}
|
|
store.add(exact, assignment_payload(scm_resume=receipt))
|
|
store.offer(0)
|
|
with store._connect() as connection:
|
|
connection.execute("UPDATE assignments SET lease_until=1")
|
|
monkeypatch.setattr(coordinator.supervisor_state, "reissue_publication_retry", lambda *_args: True)
|
|
import publication_retry_fence
|
|
marker = "[hermes-publication-retry-fence:23:" + "d" * 64 + "]"
|
|
monkeypatch.setattr(
|
|
publication_retry_fence, "block",
|
|
lambda db, connection, board, task_id, run_id, reason, kind, fenced: (
|
|
db.block_task(connection, task_id, reason=reason, kind=kind, expected_run_id=int(run_id)) or True
|
|
),
|
|
)
|
|
monkeypatch.setattr(publication_retry_fence, "recover", lambda *_args: (17, marker))
|
|
|
|
coordinator.Coordinator(MASTER, store).expire_leases()
|
|
result = store.terminal_record("metis", "t_deadbeef", "23")["result"]
|
|
assert result["scm_resume"] == receipt and result["publication_retry_lease_fence"] == marker
|
|
assert kanban.blocked[-1][1]["kind"] == "transient"
|
|
|
|
|
|
def test_crashed_fenced_lease_reconstructs_only_the_current_native_event(tmp_path, monkeypatch):
|
|
"""A native blocked/null-run shape resumes only through the preserved fence."""
|
|
blocked = task(status="blocked", current_run_id=None)
|
|
install_kanban(monkeypatch, [blocked], tmp_path)
|
|
store = pool_store.PoolStore(tmp_path / "lease-crash-fence.db")
|
|
exact = binding(attempt=3)
|
|
receipt = {
|
|
"source": {"board": "metis", "task_id": "t_deadbeef", "worker_ordinal": 0},
|
|
"result_digest": "e" * 64,
|
|
}
|
|
store.add(exact, assignment_payload(scm_resume=receipt))
|
|
with store._connect() as connection:
|
|
connection.execute("UPDATE assignments SET state='lease_failed'")
|
|
monkeypatch.setattr(coordinator.supervisor_state, "reissue_publication_retry", lambda *_args: True)
|
|
import publication_retry_fence
|
|
marker = "[hermes-publication-retry-fence:23:" + "e" * 64 + "]"
|
|
monkeypatch.setattr(publication_retry_fence, "recover", lambda *_args: (41, marker))
|
|
coordinator.Coordinator(MASTER, store).expire_leases()
|
|
assert store.terminal_record("metis", "t_deadbeef", "23")["result"]["publication_retry_lease_fence"] == marker
|
|
|
|
store = pool_store.PoolStore(tmp_path / "lease-crash-human.db")
|
|
store.add(exact, assignment_payload(scm_resume=receipt))
|
|
with store._connect() as connection:
|
|
connection.execute("UPDATE assignments SET state='lease_failed'")
|
|
monkeypatch.setattr(publication_retry_fence, "recover", lambda *_args: None)
|
|
coordinator.Coordinator(MASTER, store).expire_leases()
|
|
assert states(store) == [("23", 0, 3, "lease_failed")]
|