From 7ef0e8c788a3b16890922b7c291b1d81cce85ef7 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 3 Aug 2026 15:00:08 +0200 Subject: [PATCH] jobq: key transient tombstones per pill, not per agent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deletes `TransientState`, adds `transient_kind` to `TransientCleared`, and fixes a crash misreport — three changes over the same functions. `TransientState` was `RunningTransient` with `agent` dropped and `takes_container_down` renamed; that rename was the only thing it did, and its three consumers each read a disjoint subset. `transient_snapshot` now returns `RunningTransient` directly. `recent_transient` was keyed by agent alone and overwritten on each clear. An agent can clear several pills in one grace window, so a `Prebuild` (`takes_container_down = false`) landing after a `StopForUpdate` (`true`) left the tombstone reading `false` and the crash watcher reported a deliberate stop as a container crash. Keyed by `(agent, label)` now, with `recent_transient_within` folding back per agent by OR — the same question `crash_watch` asks of the active set. `TransientCleared` gains the label for the same reason: a client holding two open pills for one agent could not tell which one a clear referred to. The out-of-band suppression guard has no node and so no label; it uses `NO_NODE_LABEL`, angle-bracketed to stay out of the `NodeKind::as_str` namespace. --- hive-c0re/src/coordinator.rs | 141 ++++++++++++++-------- hive-c0re/src/dashboard/state_snapshot.rs | 2 +- hive-c0re/src/dashboard_events.rs | 14 ++- hive-c0re/src/job_queue/model.rs | 2 +- hive-c0re/src/job_queue/scheduler.rs | 2 +- hive-c0re/src/workers/crash_watch.rs | 2 +- 6 files changed, 110 insertions(+), 53 deletions(-) diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 8c578738..0c88b9ea 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -8,13 +8,13 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use anyhow::{Context, Result}; -use chrono::{DateTime, Utc}; use tokio::sync::{broadcast, watch}; use crate::approvals::Approvals; use crate::broker::Broker; use crate::container_view::{self, ContainerView}; use crate::dashboard_events::DashboardEvent; +use crate::job_queue::RunningTransient; use crate::operator_questions::OperatorQuestions; use crate::socket_server::{self, AgentSocket}; @@ -137,7 +137,17 @@ pub struct Coordinator { /// agents whose tombstone is still inside the grace window. Crash /// watcher consults both this and the active map before declaring /// a stop deliberate. - recent_transient: Mutex>, + /// + /// 🚨 **Keyed by `(agent, label)`, not by agent.** An agent can have several + /// pills clearing in the same window — the transient set tests status alone, + /// so a lease-exempt `Prebuild` and a lease-holding `StopForUpdate` are both + /// live and both clear. Keyed by agent, the last clear *overwrites* the + /// others: a `Prebuild` (`deliberate_stop = false`) landing after a + /// `StopForUpdate` (`true`) leaves the tombstone reading `false`, and the + /// crash watcher then reports an intentional stop as a **crash**. The + /// out-of-band suppression guard, which has no node behind it, uses + /// [`NO_NODE_LABEL`]. + recent_transient: Mutex>, /// Timestamps of recent unexpected container crashes, keyed by agent. /// Fed by `crash_watch` each time it classifies a stop as a crash (so /// a crash-looping container — which `Restart=on-failure` flips back @@ -359,38 +369,39 @@ pub struct AgentPaths { pub notes: PathBuf, } -/// Per-agent in-progress state that the dashboard surfaces between approve -/// click and container ready. +/// Collapse per-pill tombstones to one answer per agent: **was any recently +/// cleared pill for this agent a deliberate stop?** /// -/// The two fields answer genuinely different questions and are set -/// independently on purpose. There used to be a single `TransientKind` enum -/// serving both, which meant a display concern and a safety decision shared one -/// vocabulary and moved together. -#[derive(Debug, Clone)] -pub struct TransientState { - /// What the dashboard pill renders. For queue-driven work this is the - /// running node's own wire tag ([`crate::job_queue::NodeKind::as_str`]) — - /// the same vocabulary the DAG view ships, so a pill and a node name an - /// operation identically. Work with no node behind it (destroy, migration) - /// supplies its own. - /// - /// Display only. Nothing branches on it — match on a string and this - /// becomes a taxonomy again, silently. - pub label: String, - /// Whether the container going down is **expected**, i.e. this operation - /// takes it down on purpose. Read by the crash watcher to tell a - /// deliberate stop from a crash, so a wrong value here either raises a - /// false alarm or swallows a real one. - /// - /// Set by whoever creates the transient, which is the only place that - /// actually knows — it is not recoverable from `label`. - pub deliberate_stop: bool, - /// When the operation started. Wall-clock rather than `Instant` because a - /// derived entry takes it from the node's own `started_at` — the true start - /// of the work, not the moment a watcher first noticed it. - pub since: DateTime, +/// `OR`, not last-write-wins — that is the whole fix. An agent can clear several +/// pills inside one grace window (the transient set tests status alone, so a +/// lease-exempt `Prebuild` and a lease-holding `StopForUpdate` are both live), +/// and taking the last one means an incidental `false` erases a real `true`, +/// which the crash watcher then reads as a container **crash**. +/// +/// Asks the same question of the cleared set that `crash_watch` asks of the +/// active set with `.any(…)`, so the two agree by construction. A free function +/// so it is testable without a `Coordinator` fixture — same reason +/// `crash_watch::is_deliberate_stop` is one. +fn fold_tombstones_by_agent<'a>( + entries: impl Iterator, +) -> HashMap { + let mut out: HashMap = HashMap::new(); + for (agent, deliberate) in entries { + *out.entry(agent.to_owned()).or_default() |= deliberate; + } + out } +/// Tombstone label for work with **no queue node behind it** — the +/// out-of-band operations (destroy, migration) that hold a +/// [`CrashWatchGuard`] instead of appearing in the derived transient set. +/// +/// [`Coordinator::recent_transient`] is keyed by `(agent, label)` so concurrent +/// pills can't overwrite each other's `deliberate_stop`; a guard has no node and +/// therefore no node label, so it needs one of its own. Angle-bracketed to keep +/// it out of the `NodeKind::as_str` namespace — no node can ever render this. +const NO_NODE_LABEL: &str = ""; + /// RAII handle returned by [`Coordinator::suppress_crash_watch`]. While held, /// the crash watcher treats this container disappearing as **expected**. /// @@ -425,11 +436,10 @@ impl Drop for CrashWatchSuppression { // Tombstone the release so the next poll — which may land in the // window between the container going away and this guard dropping — // still reads the stop as deliberate. - self.coord - .recent_transient - .lock() - .unwrap() - .insert(self.name.clone(), (true, std::time::Instant::now())); + self.coord.recent_transient.lock().unwrap().insert( + (self.name.clone(), NO_NODE_LABEL.to_owned()), + (true, std::time::Instant::now()), + ); } } @@ -1136,14 +1146,20 @@ impl Coordinator { /// derived read of current state cannot answer "was one here a moment ago?". /// /// Old entries are reaped lazily on read, so the map stays bounded. - pub(crate) fn emit_transient_cleared(&self, name: &str, deliberate_stop: bool) { + /// `label` identifies *which* pill cleared. An agent can hold several at + /// once, so both the tombstone key and the wire event need it — without it + /// a client that received two `TransientSet`s cannot tell which one this + /// clears, and the tombstone silently overwrites a concurrent pill's + /// `deliberate_stop`. + pub(crate) fn emit_transient_cleared(&self, name: &str, label: &str, deliberate_stop: bool) { self.recent_transient.lock().unwrap().insert( - name.to_owned(), + (name.to_owned(), label.to_owned()), (deliberate_stop, std::time::Instant::now()), ); self.emit_dashboard_event(DashboardEvent::TransientCleared { seq: self.next_seq(), name: name.to_owned(), + transient_kind: label.to_owned(), }); } @@ -1222,9 +1238,7 @@ impl Coordinator { let now = std::time::Instant::now(); let mut map = self.recent_transient.lock().unwrap(); map.retain(|_, (_, ts)| now.duration_since(*ts) <= grace); - map.iter() - .map(|(k, (deliberate, _))| (k.clone(), *deliberate)) - .collect() + fold_tombstones_by_agent(map.iter().map(|((a, _), (d, _))| (a.as_str(), *d))) } /// Record an unexpected crash for `agent`. Called by the crash @@ -1301,14 +1315,10 @@ impl Coordinator { /// `deliberate_stop = true` behind a `false` — reporting an intentional /// stop as a crash. #[must_use] - pub fn transient_snapshot(&self) -> HashMap> { - let mut out: HashMap> = HashMap::new(); + 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.entry(t.agent.clone()).or_default().push(t); } out } @@ -1659,6 +1669,41 @@ pub fn rebuilt_todo_summary( summary } +#[cfg(test)] +mod tombstone_fold_tests { + use super::fold_tombstones_by_agent; + + #[test] + fn a_deliberate_stop_survives_an_incidental_clear_on_the_same_agent() { + // The regression this fold exists for. `a` clears two pills in one + // grace window: a `StopForUpdate` that really does take the container + // down, and a lease-exempt `Prebuild` that doesn't. Keyed by agent + // alone the second overwrote the first and the crash watcher reported + // a deliberate stop as a crash — order must not matter. + let out = fold_tombstones_by_agent([("a", true), ("a", false)].into_iter()); + assert_eq!(out.get("a"), Some(&true)); + + let reversed = fold_tombstones_by_agent([("a", false), ("a", true)].into_iter()); + assert_eq!(reversed.get("a"), Some(&true)); + } + + #[test] + fn all_incidental_clears_stay_false() { + // The other direction has to keep working: nothing deliberate cleared, + // so a container going down now really is a crash. + let out = fold_tombstones_by_agent([("a", false), ("a", false)].into_iter()); + assert_eq!(out.get("a"), Some(&false)); + } + + #[test] + fn agents_do_not_bleed_into_each_other() { + let out = fold_tombstones_by_agent([("a", true), ("b", false)].into_iter()); + assert_eq!(out.get("a"), Some(&true)); + assert_eq!(out.get("b"), Some(&false)); + assert_eq!(out.get("c"), None); + } +} + #[cfg(test)] mod rebuilt_todo_summary_tests { use super::rebuilt_todo_summary; diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index 78d8f822..97fb208a 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -533,7 +533,7 @@ 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() diff --git a/hive-c0re/src/dashboard_events.rs b/hive-c0re/src/dashboard_events.rs index 8c6c8db0..97cd50da 100644 --- a/hive-c0re/src/dashboard_events.rs +++ b/hive-c0re/src/dashboard_events.rs @@ -157,7 +157,18 @@ pub enum DashboardEvent { }, /// The matching lifecycle action resolved (success or failure). /// Clients drop the spinner row. - TransientCleared { seq: u64, name: String }, + /// + /// `transient_kind` matches the `TransientSet` that opened this pill, and is + /// what makes the pair addressable: an agent can hold **several** pills at + /// once (the transient set tests status alone, so a lease-exempt `Prebuild` + /// and a lease-holding `StopForUpdate` are both live). Without it a client + /// holding two open pills for one agent cannot tell which one cleared, and + /// has to drop both or guess. + TransientCleared { + seq: u64, + name: String, + transient_kind: String, + }, /// One container row changed — new container appeared (post-spawn /// finalise), an existing one flipped `running` / `needs_update` / /// `sha`, etc. Clients upsert by `container.name`. Payload carries @@ -392,6 +403,7 @@ mod tests { DashboardEvent::TransientCleared { seq: 1, name: "x".into(), + transient_kind: "rebuilding".into(), }, DashboardEvent::ContainerRemoved { seq: 1, diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 3b8aaf4b..9f45c3b1 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -389,7 +389,7 @@ impl NodeKind { } /// Whether running this node is *expected* to take the agent's container - /// down. Feeds `TransientState::deliberate_stop`, which the crash watcher + /// down. Feeds `RunningTransient::takes_container_down`, which the crash watcher /// reads to tell an intentional stop from a crash. /// /// This is a **safety** question, not a display one — it decides whether a diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index 93341414..3c19c529 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -190,7 +190,7 @@ fn reconcile_transients(coord: &Arc, prev: &mut TransientSeen) { .iter() .any(|t| &t.agent == agent && &t.label == label); if !still { - coord.emit_transient_cleared(agent, *deliberate); + coord.emit_transient_cleared(agent, label, *deliberate); } still }); diff --git a/hive-c0re/src/workers/crash_watch.rs b/hive-c0re/src/workers/crash_watch.rs index 061567d9..113072a8 100644 --- a/hive-c0re/src/workers/crash_watch.rs +++ b/hive-c0re/src/workers/crash_watch.rs @@ -97,7 +97,7 @@ fn emit_crash_transitions(coord: &Coordinator, prev: &HashSet, current: // a deliberate stop rather than a crash. let active = transients .get(stopped) - .map(|sts| sts.iter().any(|st| st.deliberate_stop)) + .map(|sts| sts.iter().any(|st| st.takes_container_down)) .or_else(|| coord.crash_watch_suppressed(stopped).then_some(true)); let recently_cleared = recent.get(stopped).copied(); if is_deliberate_stop(active, recently_cleared) {