fix(kvb-org-queue): a hotfix held back by host load holds back the entries after it

The grant order ignored live load: a prio-1 hotfix that fit the budget but waited on RAM or CPU counted as granted for the entries after it, so CPU-light jobs behind it could start, keep the load up and starve it. Such a hotfix now holds back every entry after it until it starts (operator pick 2026-11-11 on Main's report). Each holder stores the reason of its last poll, and list shows it, so a job the load holds back no longer reads as granted, without list sampling the CPU.

🎯 Fix: 92 %
💥 Break: 4 %
💸 Cost: 100 %
⏱️ Time: 100 %
🛡️ Security: 100 %
🏆 Score: 88
🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 %

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-BirkenbachandClaude Opus 5.5 committed 2026-11-11 16:34:37 +00:00
1 parent 5053c6a9be
commit 134b1d30f1
2 files changed
+55 -12

No files matched your search

+37 -12
View File
@@ -46,7 +46,8 @@ Args:
list: a header, then one line per entry: position (run for active jobs),
id, status, kind, task, need GB, minutes waited, ETA minutes (until a
running job ends or a waiting one starts, from learned medians; - while
one is unknown), workdir, blocking reason. Entries of dead holders are
one is unknown), workdir, blocking reason (the one the holder saw at
its last poll, host load included). Entries of dead holders are
dropped first.
history: a markdown table per command signature and one per kind of what
the last 30 days of history teach, with each slot limit's source.
@@ -347,7 +348,8 @@ def over_budget(used: dict[str, float], share: dict[str, float], budget: dict[st
def evaluate(entries: list[dict], free_gb: float, floor: float, stage: str, memory: str, silenced: bool,
slot_limits: dict[str, int], footprints: dict[str, dict[str, float]], budget: dict[str, float],
ranks: dict[str, int], agents: dict[str, int], now: float) -> dict[str, str | None]:
ranks: dict[str, int], agents: dict[str, int], loaded: dict[str, str],
now: float) -> dict[str, str | None]:
"""Decide every waiting entry in grant order, without disk or processes.
Args:
@@ -367,6 +369,8 @@ def evaluate(entries: list[dict], free_gb: float, floor: float, stage: str, memo
agents: {agent name: index in the operator's agent order}; it breaks
ties of equal rank only, an entry whose holder is missing there after
all of them.
loaded: {entry id: why live CPU or RAM holds it back (load_reason())}
for the waiting entries the host load keeps out.
now: epoch seconds; each whole hour waited moves an entry one rank up.
Returns:
@@ -374,8 +378,10 @@ def evaluate(entries: list[dict], free_gb: float, floor: float, stage: str, memo
An unblocked entry counts as granted for the entries after it, whoever
holds it, so a lower entry never takes a higher entry's slot; one
blocked by disk or budget reserves its need against the entries after it.
A waiting entry whose holder has not polled for CLAIM_SECONDS reserves
nothing and comes last.
A hotfix entry that passes every check but the host load holds back
every entry after it, so lighter jobs cannot keep the load up. A waiting
entry whose holder has not polled for CLAIM_SECONDS reserves nothing and
comes last.
"""
active = [e for e in entries if e["status"] == "active"]
available = free_gb - sum(e["need_gb"] for e in active)
@@ -385,12 +391,18 @@ def evaluate(entries: list[dict], free_gb: float, floor: float, stage: str, memo
busy = {e["workdir"] for e in active}
verdicts: dict[str, str | None] = {}
waiting = [e for e in entries if e["status"] == "waiting"]
held = None
for entry in ordered([e for e in waiting if not silent(e, now)], ranks, agents, now):
if held is not None:
verdicts[entry["id"]] = f"waits behind hotfix {held}, which the host load holds back"
continue
kind, need, workdir, share = entry["kind"], entry["need_gb"], entry["workdir"], footprints[entry["id"]]
reason = host_reason(stage, memory, silenced, need) or slot_reason(kind, workdir, slot_limits, counts, busy)
if reason is None:
reason, used, available = resources(need, share, used, available, floor, budget, taken)
if reason is None:
if reason is None and entry["hotfix"] and entry["id"] in loaded:
reason, held = loaded[entry["id"]], entry["id"]
elif reason is None:
taken += 1
counts[kind] += 1
busy.add(workdir)
@@ -475,8 +487,19 @@ def footprint(entry: dict, signatures: dict[str, dict], kinds: dict[str, dict])
return history.footprint(entry["kind"], signature_of(entry), signatures, kinds)
def held_back(cfg: dict, entries: list[dict], now: float, signatures: dict[str, dict], kinds: dict[str, dict],
cpu: float, mem: float) -> dict[str, str]:
loaded = {}
for entry in entries:
if entry["status"] == "waiting":
pending = pending_ram(entries, entry["id"], now, signatures, kinds)
if (reason := load_reason(cfg["load_limits"], cpu, mem, pending)) is not None:
loaded[entry["id"]] = reason
return loaded
def decide(cfg: dict, entries: list[dict], free_gb: float, now: float, signatures: dict[str, dict],
kinds: dict[str, dict]) -> dict[str, str | None]:
kinds: dict[str, dict], loaded: dict[str, str]) -> dict[str, str | None]:
slots = {kind: limit for kind, (limit, _) in limits(cfg, kinds, entries).items()}
budget = cfg["budget_percent"]
footprints = {}
@@ -486,7 +509,7 @@ def decide(cfg: dict, entries: list[dict], free_gb: float, now: float, signature
return evaluate(
entries, free_gb, cfg["thresholds_gb"]["stop"], stage(cfg, free_gb, now),
watch.memory_stage(watch.memory(), cfg), silenced(now), slots, footprints, budget,
register_ranks(), agent_ranks(), now,
register_ranks(), agent_ranks(), loaded, now,
)
@@ -591,9 +614,10 @@ def wait(cfg: dict, entry_id: str, caught: list[int]) -> tuple[int | None, int]:
print(f"queue: {entry_id} was cancelled", file=sys.stderr)
return CANCELLED, 0
rejoin(data, mine[0], now)
verdicts = decide(cfg, data["entries"], free, now, signatures, kinds)
pending = pending_ram(data["entries"], entry_id, now, signatures, kinds)
reason = verdicts[entry_id] or load_reason(cfg["load_limits"], cpu, mem, pending)
loaded = held_back(cfg, data["entries"], now, signatures, kinds, cpu, mem)
verdicts = decide(cfg, data["entries"], free, now, signatures, kinds, loaded)
reason = verdicts[entry_id] or loaded.get(entry_id)
mine[0]["reason"] = reason
if reason is None:
concurrent = sum(e["status"] == "active" for e in data["entries"])
mine[0].update(status="active", granted=now)
@@ -724,7 +748,7 @@ def cmd_list(_: argparse.Namespace) -> int:
with store() as data:
data["entries"] = prune(data["entries"])
entries = list(data["entries"])
verdicts = decide(cfg, entries, free, now, signatures, kinds)
verdicts = decide(cfg, entries, free, now, signatures, kinds, {})
by_id = {e["id"]: e for e in entries}
running = [(signature_of(e), e["granted"]) for e in entries if e["status"] == "active"]
rows = [
@@ -734,7 +758,8 @@ def cmd_list(_: argparse.Namespace) -> int:
ahead: list[str] = []
for n, (entry_id, reason) in enumerate(verdicts.items(), 1):
e = by_id[entry_id]
rows.append((str(n), e, history.eta(signatures, running, ahead, now), reason or "granted on its next poll"))
polled = reason if silent(e, now) else e.get("reason") or reason
rows.append((str(n), e, history.eta(signatures, running, ahead, now), polled or "granted on its next poll"))
ahead.append(signature_of(e))
print("position\tid\tstatus\tkind\ttask\tneed_gb\twaited_min\teta_min\tworkdir\treason")
for position, e, eta, reason in rows:
+18
View File
@@ -31,6 +31,7 @@ BASE = {
"budget": {"ram": 75, "cpu": 100},
"ranks": {"work/a": 0, "work/b": 1, "work/c": 2},
"agents": {},
"loaded": {},
"now": NOW,
}
CHILD = (
@@ -253,6 +254,16 @@ class TestEvaluate(unittest.TestCase):
])
self.assertEqual(verdicts, {"j2": "/w runs another job", "j3": None, "j4": "/x runs another job"})
def test_a_hotfix_the_host_load_holds_back_holds_back_every_entry_after_it_and_a_normal_one_does_not(self):
load = "RAM 55 % used + 42 % still to come reach the limit 90 %"
hotfix, light = job(1, hotfix="main-red"), job(2, task="work/b")
self.assertEqual(
self.decide([hotfix, light], loaded={"j1": load}),
{"j1": load, "j2": "waits behind hotfix j1, which the host load holds back"},
)
self.assertEqual(self.decide([job(1), light], loaded={"j1": load}), {"j1": None, "j2": None})
self.assertEqual(self.decide([hotfix, light]), {"j1": None, "j2": None})
class TestHistory(unittest.TestCase):
@classmethod
@@ -514,6 +525,13 @@ class TestQueueCommands(unittest.TestCase):
)
self.assertEqual(self.entries(), [])
def test_list_shows_the_host_load_the_holder_saw_at_its_last_poll(self):
self.cfg["load_limits"]["mem_percent"] = 0
self.write_config()
self.blocked_hold("true")
reason = self.run_queue("list").stdout.splitlines()[1].split("\t")[-1]
self.assertRegex(reason, r"^RAM \d+ % used \+ \d+ % still to come reach the limit 0 %$")
def test_hold_refuses_bad_input_with_exit_2_and_enqueues_nothing(self):
def hold(*args: str) -> list[str]:
return ["hold", "--session", "s1", *args]