jobq: a transient is any running node naming the agent

Per mara on #2822: status is the only test. The agent comes off the
node's own payload rather than a declared Resource::Agent edge, so the
lease-exempt kinds (Prebuild, MetaSync) that name an agent without
holding its lease now light a pill — they are work on that agent.

Dropping that test breaks the one-pill-per-agent invariant, since
lease-exemption is exactly what lets one DAG build for an agent while
another holds its lease. Everything keyed by agent alone had to follow:

- reconcile_transients keys (agent, label) via TransientSeen, so a
  second pill cannot evict the first — and cannot lose its
  takes_container_down, which the crash watcher reads at clear time.
- transient_snapshot returns a Vec per agent for the same reason. The
  collapse was silent: a Prebuild could evict a StopForUpdate and its
  deliberate_stop, making an intentional stop report as a crash.
- crash_watch asks whether ANY running node expects the container down.
- the dashboard renders one row per node instead of one per agent.

takes_container_down never reached the frontend; no wire change needed.

315 tests pass unchanged.
This commit is contained in:
atlas 2026-08-03 13:03:17 +02:00 committed by mara
commit 8ce265fdf0
6 changed files with 79 additions and 56 deletions

View file

@ -28,6 +28,17 @@ use std::sync::Arc;
use super::exec;
use crate::coordinator::Coordinator;
/// Pills published on the previous tick: `(agent, label) -> takes_container_down`.
///
/// Keyed by the **pair**, not by agent. An agent can have several pills at once
/// now that [`super::JobQueue::running_transients`] tests status alone — a
/// lease-exempt `Prebuild` for `a` runs happily while another DAG holds `a`'s
/// lease, and both name `a`. Keying by agent would drop one arbitrarily and,
/// worse, lose its `takes_container_down` — which is the crash watcher's input
/// and is stored as the value precisely so it survives to *clear* time, when the
/// node that carried it is already gone.
type TransientSeen = HashMap<(String, String), bool>;
/// Scheduler loop. Spawned once at hive-c0re startup from `main.rs`.
///
/// Shutdown semantics: subscribes to `coord.shutdown_rx()`. On a true signal
@ -50,12 +61,11 @@ use crate::coordinator::Coordinator;
/// reconverging silently.
pub async fn run_worker(coord: Arc<Coordinator>) {
let mut shutdown = coord.shutdown_rx();
// Last derived pill set we published, keyed by agent (its lease is cap-1,
// so one pill each). Purely the previous value of a *derived* quantity —
// it exists to spot transitions, since the dashboard wants edges
// (`TransientSet` / `TransientCleared`) and the crash watcher wants the
// moment of the clear. Nothing owns a pill; nothing can leak one.
let mut transients: HashMap<String, (String, bool)> = HashMap::new();
// Last derived pill set we published. Purely the previous value of a
// *derived* quantity — it exists to spot transitions, since the dashboard
// wants edges (`TransientSet` / `TransientCleared`) and the crash watcher
// wants the moment of the clear. Nothing owns a pill; nothing can leak one.
let mut transients = TransientSeen::new();
loop {
// Checked every iteration, not just in the `select!` below — a
// continuous stream of ready claims never reaches the `select!`, so
@ -168,15 +178,14 @@ pub async fn run_worker(coord: Arc<Coordinator>) {
/// starting* is a real crash that must keep reporting as one.
///
/// [`NodeKind::takes_container_down`]: super::NodeKind::takes_container_down
fn reconcile_transients(coord: &Arc<Coordinator>, prev: &mut HashMap<String, (String, bool)>) {
fn reconcile_transients(coord: &Arc<Coordinator>, prev: &mut TransientSeen) {
let running = coord.job_queue.running_transients();
// Cleared: in `prev`, gone (or relabelled) now. Emitted before the sets
// below so a same-agent label change reads as clear-then-set rather than
// two overlapping pills. `deliberate_stop` is carried in `prev` precisely
// so it is still available *here* — the node it came from is, by
// definition, no longer running to be asked.
prev.retain(|agent, (label, deliberate)| {
// Cleared: in `prev`, gone now. Emitted before the sets below so a
// replacement reads as clear-then-set rather than two overlapping pills.
// `deliberate_stop` is the value precisely so it is still available *here* —
// the node it came from is, by definition, no longer running to be asked.
prev.retain(|(agent, label), deliberate| {
let still = running
.iter()
.any(|t| &t.agent == agent && &t.label == label);
@ -187,10 +196,11 @@ fn reconcile_transients(coord: &Arc<Coordinator>, prev: &mut HashMap<String, (St
});
for t in running {
if prev.get(&t.agent).map(|(l, _)| l) == Some(&t.label) {
let key = (t.agent, t.label);
if prev.contains_key(&key) {
continue;
}
coord.emit_transient_set(&t.agent, t.label.clone());
prev.insert(t.agent, (t.label, t.takes_container_down));
coord.emit_transient_set(&key.0, key.1.clone());
prev.insert(key, t.takes_container_down);
}
}