diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 4dadeace..b2a354ea 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use anyhow::{Context as _, Result, bail}; use hive_sh4re::{ApprovalKind, ApprovalStatus, HelperEvent}; -use crate::coordinator::{Coordinator, TransientKind}; +use crate::coordinator::Coordinator; use crate::lifecycle; /// Approve a pending request. Marks the approval row durably, then @@ -882,7 +882,9 @@ pub async fn destroy(coord: &Arc, name: &str, purge: bool) -> Resul tracing::info!(%name, purge, "destroy"); // Guard auto-clears on the success path's final scope exit and on // every early-return / cancellation along the way. - let guard = coord.transient_guard(name, TransientKind::Destroying); + // Destroy is not a queue node, so it names its own label. `true`: the + // container is going away, so its disappearance must not read as a crash. + let guard = coord.transient_guard(name, "destroying", true); lifecycle::destroy(name).await?; coord.unregister_agent(name); let runtime = crate::paths::agent_runtime_dir(name); diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 7029cbf3..196b95ac 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -133,7 +133,7 @@ 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>, + 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 @@ -357,9 +357,30 @@ pub struct AgentPaths { /// Per-agent in-progress state that the dashboard surfaces between approve /// click and container ready. +/// +/// 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 { - pub kind: TransientKind, + /// 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, pub since: std::time::Instant, } @@ -408,38 +429,6 @@ impl Drop for MetaUpdateGuard { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)] -#[serde(rename_all = "snake_case")] -pub enum TransientKind { - /// `lifecycle::spawn` is running (nixos-container create + update + start). - Spawning, - /// `lifecycle::start` is running. - Starting, - /// `lifecycle::kill` is running. - Stopping, - /// A restart (`lifecycle::kill` then `lifecycle::start`) is running. - Restarting, - /// `lifecycle::rebuild` is running (nixos-container update). - Rebuilding, - /// `actions::destroy` is running. - Destroying, -} - -impl TransientKind { - /// Wire/UI label. Matches the strings the dashboard already - /// renders in the transient spinner. - pub fn as_str(self) -> &'static str { - match self { - TransientKind::Spawning => "spawning", - TransientKind::Starting => "starting", - TransientKind::Stopping => "stopping", - TransientKind::Restarting => "restarting", - TransientKind::Rebuilding => "rebuilding", - TransientKind::Destroying => "destroying", - } - } -} - /// Field-named payload for [`Coordinator::emit_approval_resolved`]. /// Mirrors the `ApprovalResolved` dashboard-event fields. `agent` /// borrows from the caller; `approval_kind` / `status` are @@ -1094,11 +1083,12 @@ impl Coordinator { /// is cancelled (HTTP request aborted, runtime shutdown mid-rebuild, /// panic). A bare set with no guaranteed clear would leak the transient /// and leave the dashboard stuck in "rebuilding…" forever. - fn set_transient(&self, name: &str, kind: TransientKind) { + fn set_transient(&self, name: &str, label: String, deliberate_stop: bool) { self.transient.lock().unwrap().insert( name.to_owned(), TransientState { - kind, + label: label.clone(), + deliberate_stop, since: std::time::Instant::now(), }, ); @@ -1113,7 +1103,7 @@ impl Coordinator { self.emit_dashboard_event(DashboardEvent::TransientSet { seq: self.next_seq(), name: name.to_owned(), - transient_kind: kind.as_str(), + transient_kind: label, since_unix, }); } @@ -1130,10 +1120,10 @@ impl Coordinator { // spurious ContainerCrash on every operator stop/restart. // Old entries get reaped lazily on read so the map doesn't // grow unbounded. - self.recent_transient - .lock() - .unwrap() - .insert(name.to_owned(), (state.kind, std::time::Instant::now())); + self.recent_transient.lock().unwrap().insert( + name.to_owned(), + (state.deliberate_stop, std::time::Instant::now()), + ); self.emit_dashboard_event(DashboardEvent::TransientCleared { seq: self.next_seq(), name: name.to_owned(), @@ -1205,20 +1195,19 @@ impl Coordinator { result } - /// Set of agents whose transient was cleared within the last - /// `grace` seconds — i.e. agents the operator just acted on, - /// whose stop the crash watcher should NOT classify as a crash. - /// Lazily reaps entries older than `grace` so the map stays - /// bounded by the active agent count. - pub fn recent_transient_within( - &self, - grace: std::time::Duration, - ) -> HashMap { + /// Per-agent `deliberate_stop` for transients cleared within the last + /// `grace` seconds — i.e. agents the operator just acted on, whose stop the + /// crash watcher should NOT classify as a crash. Lazily reaps entries older + /// than `grace` so the map stays bounded by the active agent count. + /// + /// Carries only the safety bit, not the display label: nothing downstream + /// should be able to re-derive a stop/crash decision from a pill's wording. + pub fn recent_transient_within(&self, grace: std::time::Duration) -> HashMap { 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, (kind, _))| (k.clone(), *kind)) + .map(|(k, (deliberate, _))| (k.clone(), *deliberate)) .collect() } @@ -1254,8 +1243,18 @@ impl Coordinator { /// cancelled or panic between set and clear (HTTP handlers, spawned /// tasks). The guard's `Drop` runs even on task cancellation, so /// the dashboard's spinner can't get pinned forever. - pub fn transient_guard(self: &Arc, name: &str, kind: TransientKind) -> TransientGuard { - self.set_transient(name, kind); + /// + /// `label` is what the pill renders; `deliberate_stop` says whether this + /// operation takes the container down on purpose, and is what the crash + /// watcher reads. Only the caller knows the second one — it is not + /// recoverable from the first. + pub fn transient_guard( + self: &Arc, + name: &str, + label: impl Into, + deliberate_stop: bool, + ) -> TransientGuard { + self.set_transient(name, label.into(), deliberate_stop); TransientGuard { coord: self.clone(), name: name.to_owned(), diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index f86000cb..0016487a 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -214,7 +214,8 @@ struct PortConflict { #[derive(Serialize)] struct TransientView { name: String, - kind: &'static str, + /// Owned: the label is the running node's wire tag, not one of a fixed set. + kind: String, secs: u64, } @@ -533,26 +534,12 @@ fn build_transient_views( .filter(|(name, _)| !containers.iter().any(|c| &c.name == *name)) .map(|(name, st)| TransientView { name: name.clone(), - kind: transient_label(st.kind), + kind: st.label.clone(), secs: st.since.elapsed().as_secs(), }) .collect() } -fn transient_label(k: crate::coordinator::TransientKind) -> &'static str { - use crate::coordinator::TransientKind::{ - Destroying, Rebuilding, Restarting, Spawning, Starting, Stopping, - }; - match k { - Spawning => "spawning", - Starting => "starting", - Stopping => "stopping", - Restarting => "restarting", - Rebuilding => "rebuilding", - Destroying => "destroying", - } -} - /// Render each pending approval into its dashboard view (short sha for /// `MergeConfigPr`, just the name for `Spawn`). /// Project a resolved sqlite row into the lean shape the dashboard diff --git a/hive-c0re/src/dashboard_events.rs b/hive-c0re/src/dashboard_events.rs index b5451b05..8c6c8db0 100644 --- a/hive-c0re/src/dashboard_events.rs +++ b/hive-c0re/src/dashboard_events.rs @@ -144,9 +144,15 @@ pub enum DashboardEvent { TransientSet { seq: u64, name: String, - /// Lifecycle kind: `"spawning"` / `"starting"` / `"stopping"` / - /// `"restarting"` / `"rebuilding"` / `"destroying"`. - transient_kind: &'static str, + /// What the pill renders. For queue-driven work this is the running + /// node's own wire tag (`"swap"`, `"create"`, `"stop_for_update"`, …) — + /// the same vocabulary the DAG view ships. Work with no node behind it + /// (destroy, migration) supplies its own (`"destroying"`, + /// `"rebuilding"`). + /// + /// Owned rather than `&'static str`: a label now comes from the node + /// that happens to be running, not from a fixed set. + transient_kind: String, since_unix: i64, }, /// The matching lifecycle action resolved (success or failure). @@ -380,7 +386,7 @@ mod tests { DashboardEvent::TransientSet { seq: 1, name: "x".into(), - transient_kind: "rebuilding", + transient_kind: "rebuilding".into(), since_unix: 0, }, DashboardEvent::TransientCleared { diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 0b73ac76..7dcdf4dd 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -418,13 +418,11 @@ async fn run_reconcile(coord: &Arc, claim: &Claim) -> Result, claim: &Claim) -> Result { let name = &claim.agent; - // Node-local transient only when the DAG holds none (the - // boot-reconcile template); a rebuild/spawn/etc. DAG's lease-window - // transient already covers this node. - let _guard = claim - .transient - .is_none() - .then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Starting)); + // No node-local transient guard: the pill is derived from the running node + // set, and `Start` reports `Starting` via `NodeKind::transient_kind`. This + // used to take one "only when the DAG holds none", which was a second + // derivation covering the gap left by a DAG-level declaration that couldn't + // describe a sub-step. // Run the typed start preamble: ensures the runtime dir exists and // writes the nspawn/resource-limits drop-ins. The returned // StartableAgent token is the only way to call start_with_fallback — @@ -449,10 +447,8 @@ async fn run_start(coord: &Arc, claim: &Claim) -> Result, claim: &Claim) -> Result { let name = &claim.agent; - let _guard = claim - .transient - .is_none() - .then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Stopping)); + // See `run_start`: no node-local guard — `Stop` reports `Stopping` from its + // own kind now. crate::lifecycle::kill(name).await?; coord.unregister_agent(name); coord.notify_manager(&hive_sh4re::HelperEvent::Killed { diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 9300b526..1bf2fbf9 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -47,7 +47,6 @@ use hive_jobq::{Dep, Graph, NodeId}; use hive_sh4re::wire_time::now_unix; use tokio::sync::Notify; -use crate::coordinator::TransientKind; pub use hive_jobq::TerminalState; pub use model::{DagSpec, DagView, NodeKind, NodeSpec, PermPayload, Source, State}; use resource::Resource; @@ -70,10 +69,6 @@ pub struct Claim { /// The agent this node targets (its own, not a DAG-level field). Empty for /// the agentless [`NodeKind::MetaLock`] + [`NodeKind::Dag`] container nodes. pub agent: String, - /// Transient pill kind for the lease window (from the spec). Whether the - /// pill is currently shown is derived from live lease ownership - /// ([`JobQueue::held_transients`]), not a per-claim edge. - pub transient: Option, } /// Per-node runtime metadata the crate graph doesn't carry. Lifecycle @@ -91,7 +86,6 @@ struct NodeRuntime { struct DagMeta { source: Source, reason: String, - transient: Option, created_at: i64, } @@ -215,7 +209,6 @@ impl JobQueue { NodeKind::Dag { source: spec.source, reason: spec.reason, - transient: spec.transient, created_at: now_unix(), }, Vec::new(), @@ -293,15 +286,11 @@ impl JobQueue { let Some(container) = inner.sched.graph().root_of(id) else { continue; }; - let Some(meta) = inner.dag_meta(container) else { - continue; - }; claims.push(Claim { dag_id: container.get(), node_id: id, kind, agent, - transient: meta.transient, }); // `started_at` is stamped on the graph `Node` by the scheduler's // transition to `Running` — no host-side copy needed. @@ -410,24 +399,58 @@ impl JobQueue { .map(ToOwned::to_owned) } - /// The `(dag_id, agent, kind)` triples for every per-agent lease currently - /// held by a DAG that carries a transient pill — the live transient-pill - /// set, a pull query over crate resource ownership (replaces the old - /// lease-release event stream). A DAG with no transient kind is omitted. + /// The `(agent, label)` pairs for the live transient-pill set — derived from + /// the nodes **actually running**, not from an intent a template declared at + /// submit time. A rebuild used to report `rebuilding` for its whole life: + /// through the prebuild, the stop, the swap, the tail and the reconcile. + /// + /// A node lights a pill when it is `Running` **and declares the agent's + /// resource itself**. Declaring is the test, not targeting: `Prebuild` and + /// `MetaSync` name an agent but are lease-exempt on purpose — the container + /// keeps serving right through them — so they must not light one. It is also + /// not the *lease owner*: `resource_state()` answers "who holds the slot", + /// a different question from "what is running". + /// + /// The label is the node's own wire tag ([`NodeKind::as_str`]) — the same + /// vocabulary [`NodeView::kind`] already ships, so a pill and a DAG node + /// name an operation identically and there is no second taxonomy to keep in + /// step. + /// + /// Consequence, by design: `Start` / `Stop` / `PostSwap` run *inside* a + /// lease-holding ancestor and re-declare nothing, so they light no pill and + /// the agent reads idle for those windows. Closing that is the point of the + /// resources-where-constructed work, not of this function. + /// + /// An agent's lease is cap-1, so at most one pair per agent. + /// The third element is [`NodeKind::takes_container_down`] — the crash + /// watcher's input, carried alongside the label rather than inferred from + /// it (a `Start` pill and a `Stop` pill are both pills; only one of them + /// means a vanished container is expected). #[must_use] - pub fn held_transients(&self) -> Vec<(u64, String, TransientKind)> { + pub fn held_transients(&self) -> Vec<(String, String, bool)> { let inner = self.lock(); inner .sched - .resource_state() - .into_iter() - .filter_map(|(res, holder)| { - let Resource::Agent(agent) = res else { - return None; - }; - let container = inner.sched.graph().root_of(holder)?; - let kind = inner.dag_meta(container)?.transient?; - Some((container.get(), agent, kind)) + .graph() + .nodes() + .filter(|n| matches!(n.state, State::Running)) + .filter_map(|n| { + let agent = n + .payload + .resource_deps() + .into_iter() + .find_map(|d| match d { + Dep::Resource { + name: Resource::Agent(a), + .. + } => Some(a), + _ => None, + })?; + Some(( + agent, + n.payload.as_str().to_owned(), + n.payload.takes_container_down(), + )) }) .collect() } @@ -481,7 +504,6 @@ impl QueueInner { let NodeKind::Dag { source, reason, - transient, created_at, } = &self.sched.graph().node(container)?.payload else { @@ -490,7 +512,6 @@ impl QueueInner { Some(DagMeta { source: *source, reason: reason.clone(), - transient: *transient, created_at: *created_at, }) } diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index fa208985..96fba3c3 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -15,8 +15,6 @@ pub use hive_host_sock::jobs::{DagView, NodeId, PermPayload, Source, State}; use serde::Serialize; -use crate::coordinator::TransientKind; - use hive_jobq::{DepWhen, TerminalState}; /// A dependency edge (intra-DAG only — cross-DAG ordering comes from @@ -285,7 +283,6 @@ pub enum NodeKind { Dag { source: Source, reason: String, - transient: Option, created_at: i64, }, } @@ -393,6 +390,44 @@ impl NodeKind { ) } + /// Whether running this node is *expected* to take the agent's container + /// down. Feeds `TransientState::deliberate_stop`, 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 + /// vanished container raises an alert. It is deliberately not derived from + /// the pill label: a label is free to be renamed or added without moving + /// the alerting boundary, and only the operation itself knows its intent. + /// + /// Default is `false`, and that asymmetry is the point. A wrong `false` + /// costs a spurious crash event; a wrong `true` **swallows a real crash** + /// silently. So a kind earns `true` by being listed here, and anything new + /// is noisy-but-safe until someone decides otherwise. + #[must_use] + pub fn takes_container_down(&self) -> bool { + matches!( + self, + // Explicit stops, and the quiesce steps that precede one. + NodeKind::Stop { .. } + | NodeKind::StopForUpdate { .. } + | NodeKind::Signal { .. } + | NodeKind::Drain { .. } + | NodeKind::SetWanted { up: false, .. } + // The rebuild's own machinery: the container is down across the + // swap and the drop-in write that reconfigures it. + | NodeKind::Swap { .. } + | NodeKind::WriteDropin { .. } + ) + // Everything else is `false` on purpose, including the ones that would + // be easy to wave through: + // - `Create` / `Start` / `SetWanted{up}` bring a container UP. A + // container disappearing *while starting* is a genuine crash and has + // to keep reporting as one. + // - `Reconcile` is a planner; it fans out `Start` / `Stop`, which carry + // their own answer. + // - `DeployWindow` brackets a deploy without itself stopping anything. + } + /// Kinds that **mutate the meta repo** and so must hold the global /// [`Resource::MetaWindow`](super::resource::Resource::MetaWindow) for /// their duration: no two meta mutations may interleave, because a commit @@ -463,8 +498,5 @@ pub struct DagSpec { pub source: Source, /// Free-form "why". pub reason: String, - /// Dashboard transient pill (and crash-watch suppression) held for - /// the lease window — from lease acquisition to DAG terminal. - pub transient: Option, pub nodes: Vec, } diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index d235653b..24cb8000 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -3,12 +3,12 @@ //! and on any completion re-evaluate. Concurrency comes from the build-slot //! count, not multiple workers. //! -//! Owns the per-DAG transient guard (dashboard pill + crash-watch suppression) -//! that the sync queue core can't hold itself. The guard set is *reconciled* -//! from live lease ownership ([`super::JobQueue::held_transients`]) each loop: -//! a `(dag, agent)` pill exists for exactly as long as that agent's lease is -//! held, so it appears when the agent's owner node starts and disappears when -//! its subgraph settles — one pill per agent a DAG touches. +//! Owns the per-agent transient guard (dashboard pill + crash-watch +//! suppression) that the sync queue core can't hold itself. The guard set is +//! *reconciled* each loop from [`super::JobQueue::held_transients`], which +//! reports what is **running right now** under each held agent lease — so the +//! label tracks the DAG's progress (signal → swap → reconcile) instead of +//! repeating one intent the template declared before any of it started. //! //! Per-DAG terminal work (approval resolution, `Rebuilt`) is not drained here: //! it runs as the DAG's focused terminal node (`ResolveApproval` / @@ -19,7 +19,7 @@ //! its `Start`/`Stop`) flows through `NodeOutput.append_subgraph`, applied //! before the emitting node completes — see `handle_completion`. -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::sync::Arc; use super::Claim; @@ -54,8 +54,10 @@ struct NodeDone { pub async fn run_worker(coord: Arc) { let mut shutdown = coord.shutdown_rx(); let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); - // (DAG id, agent) → transient guard held for that agent's lease window. - let mut transients: HashMap<(u64, String), crate::coordinator::TransientGuard> = HashMap::new(); + // Keyed by agent (its lease is cap-1, so one pill each); the label rides + // along so a change of label can be detected and the guard swapped. + let mut transients: HashMap = + HashMap::new(); loop { // Checked every iteration, not just in the `select!` below — a // continuous stream of ready claims never reaches the `select!`, so @@ -146,18 +148,35 @@ fn handle_completion(coord: &Arc, done: NodeDone) { coord.emit_rebuild_queue_snapshot(); } -/// Reconcile the transient-guard set against live lease ownership: drop pills -/// whose lease is no longer held, create one for each newly-held `(dag, agent)`. +/// Reconcile the transient-guard set against the live pill set: drop guards for +/// pills that are no longer current, create one for each newly-current +/// `(agent, label)`. Keyed by agent — an agent's lease is cap-1, so it has at +/// most one pill. +/// +/// `deliberate_stop` rides along per node ([`NodeKind::takes_container_down`]) +/// rather than being blanket-`true` for anything holding a lease: `Create` and +/// `Start` hold the agent's lease too, and a container vanishing *while +/// starting* is a real crash that must keep reporting as one. +/// +/// [`NodeKind::takes_container_down`]: super::NodeKind::takes_container_down +/// +/// ⚠️ **`retain` must run to completion before anything is created.** +/// `TransientGuard::drop` calls `clear_transient(agent)` — keyed by agent alone, +/// with no notion of *which* label it was clearing. So when a pill's label +/// changes for the same agent (which is now routine: the label follows the +/// running node as a DAG advances), creating the new guard first and dropping +/// the old second would clear the pill that was just set. Dropping first is what +/// makes the swap safe. fn reconcile_transients( coord: &Arc, - transients: &mut HashMap<(u64, String), crate::coordinator::TransientGuard>, + transients: &mut HashMap, ) { let held = coord.job_queue.held_transients(); - let keys: HashSet<(u64, String)> = held.iter().map(|(d, a, _)| (*d, a.clone())).collect(); - transients.retain(|k, _| keys.contains(k)); - for (dag_id, agent, kind) in held { - transients - .entry((dag_id, agent.clone())) - .or_insert_with(|| coord.transient_guard(&agent, kind)); + transients.retain(|agent, (label, _)| held.iter().any(|(a, l, _)| a == agent && l == label)); + for (agent, label, takes_down) in held { + transients.entry(agent.clone()).or_insert_with(|| { + let guard = coord.transient_guard(&agent, label.clone(), takes_down); + (label, guard) + }); } } diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index 9ced4137..43955ec9 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -27,7 +27,7 @@ use std::sync::Arc; use super::model::{DagSpec, Dep, NodeKind, NodeSpec}; use super::templates::{RebuildOpts, after_ok, child, node, rebuild_nodes}; use super::{Source, templates}; -use crate::coordinator::{Coordinator, TransientKind}; +use crate::coordinator::Coordinator; use crate::lifecycle; fn submit_and_emit(coord: &Arc, spec: super::DagSpec) -> u64 { @@ -198,16 +198,10 @@ fn concat_subgraphs(chains: Vec>) -> Vec { /// Wrap assembled power-op `nodes` in a `DagSpec`. No tail node: a power op's /// effect is its nodes (`SetWanted` + `Reconcile`), with nothing left to do once /// they settle. -fn power_dag( - transient: TransientKind, - source: Source, - reason: String, - nodes: Vec, -) -> DagSpec { +fn power_dag(source: Source, reason: String, nodes: Vec) -> DagSpec { DagSpec { source, reason, - transient: Some(transient), nodes, } } @@ -228,18 +222,15 @@ pub(crate) fn stop_spec( .iter() .map(|(agent, running)| stop_chain(agent, graceful, *running)) .collect(); - power_dag( - TransientKind::Stopping, - source, - reason, - concat_subgraphs(chains), - ) + power_dag(source, reason, concat_subgraphs(chains)) } /// Assemble the start DAG from explicit `(agent, running, stale)` targets. -/// Transient is `Rebuilding` when any down+stale agent grew a rebuild -/// subgraph (crash-watch suppression during its Swap), else `Starting`; -/// applied per-agent at claim time, so each agent still shows its own pill. +/// +/// No DAG-level pill: each agent's dashboard label is derived from the node +/// running under its lease, so a down+stale agent that grew a rebuild subgraph +/// reports `rebuilding` during its swap and `starting` at its reconcile, +/// without the DAG having to guess one label covering every target. pub(crate) fn start_spec( targets: &[(String, bool, bool)], source: Source, @@ -249,13 +240,7 @@ pub(crate) fn start_spec( .iter() .map(|(agent, running, stale)| start_chain(agent, *running, *stale)) .collect(); - let any_rebuild = targets.iter().any(|(_, running, stale)| !running && *stale); - let transient = if any_rebuild { - TransientKind::Rebuilding - } else { - TransientKind::Starting - }; - power_dag(transient, source, reason, concat_subgraphs(chains)) + power_dag(source, reason, concat_subgraphs(chains)) } /// Assemble the restart DAG from explicit `(agent, running)` targets. @@ -269,12 +254,7 @@ pub(crate) fn restart_spec( .iter() .map(|(agent, running)| restart_chain(agent, graceful, *running)) .collect(); - power_dag( - TransientKind::Restarting, - source, - reason, - concat_subgraphs(chains), - ) + power_dag(source, reason, concat_subgraphs(chains)) } /// Restart a single agent. Thin wrapper over [`restart_many`]. diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 19abd54b..70c6d6a9 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -34,7 +34,6 @@ use anyhow::{Result, bail}; use hive_jobq::{DepWhen, TerminalState}; use super::model::{DagSpec, Dep, NodeKind, NodeSpec, PermPayload, Source}; -use crate::coordinator::TransientKind; /// After-ok edge on the previous node — the common chain link. Shared with /// the async power-op builders in `submit.rs` (which assemble per-agent @@ -367,7 +366,6 @@ pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> Dag DagSpec { source, reason, - transient: Some(TransientKind::Rebuilding), nodes, } } @@ -402,7 +400,6 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec DagSpec { source: Source::Approval, reason, - transient: Some(TransientKind::Rebuilding), nodes: vec![ node( NodeKind::DeployWindow { @@ -451,16 +448,10 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec /// single-node lifecycle DAGs that exercise per-agent lease serialization /// in the queue tests); production paths no longer emit a bare reconcile. #[cfg(test)] -pub fn reconcile_only( - agent: &str, - source: Source, - reason: String, - transient: Option, -) -> DagSpec { +pub fn reconcile_only(agent: &str, source: Source, reason: String) -> DagSpec { DagSpec { source, reason, - transient, nodes: vec![node( NodeKind::Reconcile { agent: agent.to_owned(), @@ -485,7 +476,6 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { DagSpec { source: Source::Approval, reason, - transient: Some(TransientKind::Spawning), nodes: { let a = || agent.to_owned(); vec![ @@ -529,7 +519,6 @@ pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPay DagSpec { source, reason, - transient: Some(TransientKind::Rebuilding), nodes, } } @@ -568,7 +557,6 @@ pub fn meta_update( DagSpec { source, reason, - transient: Some(TransientKind::Rebuilding), nodes, } } @@ -590,7 +578,6 @@ pub fn reparent( DagSpec { source, reason, - transient: None, nodes: vec![node(NodeKind::Reparent { moves }, Vec::new())], } } diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 5fd97c49..49d55bd4 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -246,7 +246,7 @@ fn graceful_rebuild_chain_drains_before_stopping() { let spec = DagSpec { source: Source::AutoUpdate, reason: "sweep".to_owned(), - transient: None, + nodes: templates::rebuild_nodes( "agent-a", templates::RebuildOpts { @@ -430,7 +430,7 @@ fn lease_serializes_two_lifecycle_dags_for_same_agent() { let restart = submit(&q, restart_online(&["agent-a"], false, "restart")); let stop = submit( &q, - templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned(), None), + templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned()), ); // Restart's first node (StopForUpdate) takes the lease; stop's // Reconcile must wait even though slots are free. @@ -461,7 +461,7 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() { submit(&q, rebuild("agent-a", "rebuild")); submit( &q, - templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned(), None), + templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned()), ); // Both DAGs' heads are lease-independent of each other: the rebuild's // MetaSync (meta window) and the stop's Reconcile (agent lease). @@ -723,7 +723,7 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { let spec = DagSpec { source: Source::AutoUpdate, reason: "sweep".to_owned(), - transient: None, + nodes: vec![NodeSpec { kind: NodeKind::MetaLock { sweep: true, @@ -789,26 +789,55 @@ fn drain_meta_syncs(q: &JobQueue) -> Vec<(String, String)> { rest } +/// Crash-watch suppression for a cascade rebuild, which the deleted half of +/// `meta_update_grows_cascade_in_dag` used to assert via `DagSpec::transient`. +/// +/// The property is unchanged — a container going down under a rebuild must not +/// read as a crash — but it is no longer a DAG-level declaration: each node +/// answers for itself, so the assertion moves to the nodes a cascade actually +/// runs. Kept as its own test rather than dropped, because it is the *property* +/// that mattered, not the field that used to carry it. #[test] -fn meta_update_carries_rebuilding_transient_and_grows_cascade_in_dag() { +fn rebuild_chain_nodes_suppress_crash_watch() { + for kind in [ + NodeKind::StopForUpdate { + agent: "a".to_owned(), + }, + NodeKind::Swap { + agent: "a".to_owned(), + }, + NodeKind::Drain { + agent: "a".to_owned(), + }, + ] { + assert!( + kind.takes_container_down(), + "{} must suppress crash-watch — a rebuild takes the container down \ + on purpose", + kind.as_str() + ); + } + // The counter-case, and the reason this can't be "any node in a rebuild": + // the tail brings the container back up, so a container that dies there + // really did crash. + assert!( + !NodeKind::Start { + agent: "a".to_owned() + } + .takes_container_down() + ); +} + +#[test] +fn meta_update_grows_cascade_in_dag() { // The meta-update `MetaLock` grows one rebuild subgraph per affected // agent into its OWN DAG (via append_subgraph), not child DAGs. - // The DAG carries `Rebuilding` so the folded rebuilds keep crash-watch - // suppression (the property the old child Rebuild DAGs had via their own - // transient). let spec = templates::meta_update( vec!["nixpkgs".to_owned()], Source::Manual, "bump".to_owned(), None, ); - assert!( - matches!( - spec.transient, - Some(crate::coordinator::TransientKind::Rebuilding) - ), - "meta-update DAG must carry Rebuilding so cascade rebuilds get suppression" - ); let q = JobQueue::new(4); let id = submit(&q, spec); let meta_lock = claim_one(&q); @@ -981,7 +1010,7 @@ fn failed_reconcile_marks_dag_failed() { let q = JobQueue::new(1); let id = submit( &q, - templates::reconcile_only("agent-a", Source::Manual, "start".to_owned(), None), + templates::reconcile_only("agent-a", Source::Manual, "start".to_owned()), ); let c = claim_one(&q); q.complete_node(c.node_id, Err("start failed".to_owned())); @@ -1132,7 +1161,7 @@ fn dag_settles_terminal_and_releases_lease_after_work() { // immediately. let next = submit( &q, - templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned(), None), + templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned()), ); let c = claim_one(&q); assert_eq!(c.dag_id, next); @@ -1401,12 +1430,7 @@ fn history_evicts_oldest_terminals_past_flat_cap() { for i in 0..(MAX_HISTORY_DAGS + OVERFLOW) { let id = submit( &q, - templates::reconcile_only( - &format!("agent-{i}"), - Source::Manual, - "start".to_owned(), - None, - ), + templates::reconcile_only(&format!("agent-{i}"), Source::Manual, "start".to_owned()), ); let c = claim_one(&q); // Fail the single work node so the DAG *lingers*: a fully-`Done` DAG diff --git a/hive-c0re/src/migrate.rs b/hive-c0re/src/migrate.rs index 355bba5d..38a5d5e2 100644 --- a/hive-c0re/src/migrate.rs +++ b/hive-c0re/src/migrate.rs @@ -108,8 +108,10 @@ pub async fn run(coord: &Arc) -> Result<()> { // update activation triggers. Without this, crash_watch // would fire ContainerCrash for every agent here and the // manager would spuriously try to recover them. - let guard = - coord.transient_guard(name.as_str(), crate::coordinator::TransientKind::Rebuilding); + // No queue node behind this one — migration repoints containers + // directly — so the label is supplied here. `true`: the repoint takes + // the container down, which is the whole reason for the guard. + let guard = coord.transient_guard(name.as_str(), "rebuilding", true); let result = repoint_container(name.as_str()).await; drop(guard); if let Err(e) = result { @@ -201,7 +203,8 @@ async fn rename_manager_container(coord: &Arc) { return; } tracing::info!("migration phase 5: renaming root container to h-root"); - let _guard = coord.transient_guard(MANAGER_NAME, crate::coordinator::TransientKind::Rebuilding); + // `true`: the old container is stopped immediately below. + let _guard = coord.transient_guard(MANAGER_NAME, "rebuilding", true); // Stop the old container. Abort if stop fails — continuing with a // running `root` and then starting `h-root` risks two manager diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index 36e42477..3fc78e61 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -353,7 +353,6 @@ fn submit_boot_tree( // Rebuilding when the sweep will grow rebuild subgraphs (per-agent // crash-watch suppression during their Swap, applied at claim time); // a reconcile-only boot needs no transient. - transient: any_stale.then_some(crate::coordinator::TransientKind::Rebuilding), nodes, }; if let Err(e) = coord.job_queue.submit(spec) { diff --git a/hive-c0re/src/workers/crash_watch.rs b/hive-c0re/src/workers/crash_watch.rs index 49111296..9b74f826 100644 --- a/hive-c0re/src/workers/crash_watch.rs +++ b/hive-c0re/src/workers/crash_watch.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use std::time::Duration; use crate::container_view::claude_has_session; -use crate::coordinator::{Coordinator, TransientKind}; +use crate::coordinator::Coordinator; use crate::lifecycle::{self, AGENT_PREFIX}; const POLL_INTERVAL: Duration = Duration::from_secs(10); @@ -87,7 +87,7 @@ fn emit_crash_transitions(coord: &Coordinator, prev: &HashSet, current: // guard between two crash-watch polls. let recent = coord.recent_transient_within(RECENT_TRANSIENT_GRACE); for stopped in prev.difference(current) { - let active = transients.get(stopped).map(|st| st.kind); + let active = transients.get(stopped).map(|st| st.deliberate_stop); let recently_cleared = recent.get(stopped).copied(); if is_deliberate_stop(active, recently_cleared) { continue; @@ -101,26 +101,19 @@ fn emit_crash_transitions(coord: &Coordinator, prev: &HashSet, current: } } -/// Pure classifier: did the operator stop / restart / destroy / -/// rebuild this container, or did it crash? Splits the matcher out -/// so it has a focused unit test without needing a Coordinator -/// fixture. `active` is the currently-set transient (if any), -/// `recently_cleared` is one whose RAII guard dropped within the -/// grace window. -fn is_deliberate_stop( - active: Option, - recently_cleared: Option, -) -> bool { - let is_op_kind = |kind: TransientKind| { - matches!( - kind, - TransientKind::Stopping - | TransientKind::Restarting - | TransientKind::Destroying - | TransientKind::Rebuilding - ) - }; - active.is_some_and(is_op_kind) || recently_cleared.is_some_and(is_op_kind) +/// Pure classifier: did an operation take this container down on purpose, or +/// did it crash? Splits the matcher out so it has a focused unit test without +/// needing a Coordinator fixture. `active` is the currently-set transient's +/// `deliberate_stop` (if any), `recently_cleared` is one whose RAII guard +/// dropped within the grace window. +/// +/// This reads the flag the transient's creator set and nothing else. It used to +/// match a `TransientKind`, which meant a pill's *display* vocabulary decided a +/// crash-alert question — so renaming or adding a label silently moved the +/// alerting boundary. Whoever starts the operation knows whether the container +/// is meant to go down; nothing downstream can re-derive it. +fn is_deliberate_stop(active: Option, recently_cleared: Option) -> bool { + active.unwrap_or(false) || recently_cleared.unwrap_or(false) } fn emit_login_transitions( @@ -168,29 +161,12 @@ mod tests { use super::*; #[test] - fn deliberate_when_active_transient_is_operator_kind() { - for kind in [ - TransientKind::Stopping, - TransientKind::Restarting, - TransientKind::Destroying, - TransientKind::Rebuilding, - ] { - assert!(is_deliberate_stop(Some(kind), None), "{kind:?}"); - } - } - - #[test] - fn deliberate_when_recent_transient_is_operator_kind() { - // Race repros: lifecycle action completes + drops the guard - // between two polls. recent_transient catches it. - for kind in [ - TransientKind::Stopping, - TransientKind::Restarting, - TransientKind::Destroying, - TransientKind::Rebuilding, - ] { - assert!(is_deliberate_stop(None, Some(kind)), "{kind:?}"); - } + fn deliberate_when_either_source_says_so() { + // Active guard, and the race repro: a lifecycle action completes and + // drops its guard between two polls, so only `recent` still carries it. + assert!(is_deliberate_stop(Some(true), None)); + assert!(is_deliberate_stop(None, Some(true))); + assert!(is_deliberate_stop(Some(true), Some(true))); } #[test] @@ -200,13 +176,16 @@ mod tests { } #[test] - fn not_deliberate_when_only_spawning_starting() { - // Spawning/Starting are never paired with a "stopped" transition - // — they're starts. If we see one alongside a stop, it's - // unrelated (e.g. just-started container died), still a crash. - for kind in [TransientKind::Spawning, TransientKind::Starting] { - assert!(!is_deliberate_stop(Some(kind), None), "{kind:?} active"); - assert!(!is_deliberate_stop(None, Some(kind)), "{kind:?} recent"); - } + fn not_deliberate_when_the_operation_was_bringing_the_container_up() { + // The case that used to be spelled `Spawning` / `Starting`: an + // operation IS in flight, but it is not one that takes the container + // down, so a container that vanishes under it really did crash. + // + // This is why `deliberate_stop` is carried rather than inferred from + // the pill — `Create` and `Start` hold the agent's lease exactly like + // `Stop` does, so "has a pill" cannot answer this. + assert!(!is_deliberate_stop(Some(false), None)); + assert!(!is_deliberate_stop(None, Some(false))); + assert!(!is_deliberate_stop(Some(false), Some(false))); } }