From 8ce265fdf032894c67f20784f48a9a45a94f4f23 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 3 Aug 2026 13:03:17 +0200 Subject: [PATCH] jobq: a transient is any running node naming the agent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- hive-c0re/src/coordinator.rs | 31 +++++++------- hive-c0re/src/dashboard/state_snapshot.rs | 5 ++- hive-c0re/src/dashboard/tombstones.rs | 2 +- hive-c0re/src/job_queue/mod.rs | 49 +++++++++++++---------- hive-c0re/src/job_queue/scheduler.rs | 42 +++++++++++-------- hive-c0re/src/workers/crash_watch.rs | 6 ++- 6 files changed, 79 insertions(+), 56 deletions(-) diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 4998e441..8c578738 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -1294,22 +1294,23 @@ impl Coordinator { /// crash-watch suppression is a separate, narrower thing /// ([`Coordinator::suppress_crash_watch`]); the pill comes back for free /// once those become real nodes. + /// ⚠️ **A `Vec` per agent, not one entry.** `running_transients` tests + /// status alone, so a lease-exempt `Prebuild` for `a` and a lease-holding + /// `StopForUpdate` for `a` are both live pills. Collapsing them to one + /// would pick arbitrarily and, for the crash watcher, silently lose a + /// `deliberate_stop = true` behind a `false` — reporting an intentional + /// stop as a crash. #[must_use] - pub fn transient_snapshot(&self) -> HashMap { - self.job_queue - .running_transients() - .into_iter() - .map(|t| { - ( - t.agent, - TransientState { - label: t.label, - deliberate_stop: t.takes_container_down, - since: t.since, - }, - ) - }) - .collect() + pub fn transient_snapshot(&self) -> HashMap> { + let mut out: HashMap> = HashMap::new(); + for t in self.job_queue.running_transients() { + out.entry(t.agent).or_default().push(TransientState { + label: t.label, + deliberate_stop: t.takes_container_down, + since: t.since, + }); + } + out } /// Drop a system message into the given agent's inbox. Wakes the diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index 8b4c5d1a..96e9acab 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -533,11 +533,14 @@ fn build_port_conflicts(containers: &[ContainerView]) -> Vec { /// `ContainerView.pending` inline; this list only catches pre-creation. fn build_transient_views( containers: &[ContainerView], - transient_snapshot: &std::collections::HashMap, + transient_snapshot: &std::collections::HashMap>, ) -> Vec { transient_snapshot .iter() .filter(|(name, _)| !containers.iter().any(|c| &c.name == *name)) + // One row per running node, so an agent with several shows several + // rather than one of them arbitrarily. + .flat_map(|(name, sts)| sts.iter().map(move |st| (name, st))) .map(|(name, st)| TransientView { name: name.clone(), kind: st.label.clone(), diff --git a/hive-c0re/src/dashboard/tombstones.rs b/hive-c0re/src/dashboard/tombstones.rs index e432e31f..cb54d1ba 100644 --- a/hive-c0re/src/dashboard/tombstones.rs +++ b/hive-c0re/src/dashboard/tombstones.rs @@ -35,7 +35,7 @@ pub struct TombstoneView { pub(super) fn build_tombstone_views( coord: &Coordinator, containers: &[ContainerView], - transient_snapshot: &std::collections::HashMap, + transient_snapshot: &std::collections::HashMap>, ) -> Vec { let _ = coord; // kept_state_names is a free fn but takes &self by future plan let live: std::collections::HashSet<&str> = containers diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 843b6d3c..12b7c08e 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -306,26 +306,26 @@ impl JobQueue { /// template declared at submit time. (A rebuild used to report `rebuilding` /// for its whole life: prebuild, stop, swap, tail and reconcile alike.) /// - /// A node lights a pill when it is `Running` **and declares the agent's - /// resource itself**. Declaring is the test, not targeting — `Prebuild` / - /// `MetaSync` name an agent but are lease-exempt on purpose, since the - /// container keeps serving through them. Nor is it the lease *owner*: - /// `resource_state()` answers "who holds the slot", a different question. + /// **Status is the only test**: every `Running` node that names an agent is + /// in the set. Naming is targeting, not lease-holding — `Prebuild` / + /// `MetaSync` are lease-exempt (the container keeps serving through them) + /// but they *are* work on that agent, and the operator wants to see it. + /// + /// ⚠️ **So there can be more than one entry per agent**, which is the whole + /// difference from the older lease-declaration test: lease-exemption is + /// exactly what lets one DAG build for `a` while another holds `a`'s lease, + /// so both are running and both name `a`. Anything keying this set by agent + /// alone will silently drop one — see [`super::scheduler`]. /// /// `label` is the node's own wire tag ([`NodeKind::as_str`]), the vocabulary /// [`NodeView::kind`] already ships, so a pill and a DAG node name an /// operation identically. `takes_container_down` is the crash watcher's - /// input, carried rather than inferred from the label — a `Start` pill and a - /// `Stop` pill are both pills; only one means a vanished container is + /// input and does **not** ride the wire to the frontend — a `Start` pill and + /// a `Stop` pill are both pills; only one means a vanished container is /// expected. /// - /// Read off the node's **declared** resource edges, not off its kind. Those - /// are the same thing now that every construction site states what it holds, - /// and the distinction is the whole point: `Start` / `Stop` / `PostSwap` run - /// inside a lease-holding ancestor, and while the declaration was derived - /// from the kind they re-declared nothing and lit no pill. Asking the node - /// what it holds cannot go stale that way. An agent's lease is cap-1, so at - /// most one entry per agent. + /// Not the lease *owner* either: `resource_state()` answers "who holds the + /// slot", a different question. #[must_use] pub fn running_transients(&self) -> Vec { let inner = self.lock(); @@ -334,15 +334,20 @@ impl JobQueue { .nodes() .filter(|n| matches!(n.state, State::Running)) .filter_map(|n| { - let agent = n.deps.iter().find_map(|dep| match dep { - hive_jobq::Dep::Resource { - name: Resource::Agent(a), - .. - } => Some(a.clone()), - _ => None, - })?; + // Status is the only test. The agent comes off the node's own + // payload, not off a declared `Resource::Agent` edge: the + // lease-exempt kinds (`Prebuild` / `MetaSync`) name an agent + // without declaring its lease, and they are work on that agent + // that the operator wants to see. + // + // Empty means an agentless container kind (`MetaLock`, `Dag`), + // which targets no agent and lights nothing. + let agent = n.payload.agent(); + if agent.is_empty() { + return None; + } Some(RunningTransient { - agent, + agent: agent.to_owned(), label: n.payload.as_str().to_owned(), takes_container_down: n.payload.takes_container_down(), // `started_at` is set when a node enters `Running`, and this diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index 897f41ad..3dc25de1 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -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) { 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 = 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) { /// 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, prev: &mut HashMap) { +fn reconcile_transients(coord: &Arc, 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, prev: &mut HashMap, current: // Two sources, because a container can go down on purpose either way: // a running queue node that declared it takes the container down, or a // no-node operation (destroy, migration) holding a suppression guard. + // `any`, not "the" pill: an agent can have several running nodes at + // once (a lease-exempt build alongside a lease-holding stop), and it + // only takes one of them expecting the container down for this to be + // a deliberate stop rather than a crash. let active = transients .get(stopped) - .map(|st| st.deliberate_stop) + .map(|sts| sts.iter().any(|st| st.deliberate_stop)) .or_else(|| coord.crash_watch_suppressed(stopped).then_some(true)); let recently_cleared = recent.get(stopped).copied(); if is_deliberate_stop(active, recently_cleared) {