diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 7325d92e..ade13914 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -8,6 +8,7 @@ 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; @@ -118,7 +119,9 @@ pub struct Coordinator { /// Agents whose lifecycle action (currently just spawn) is in flight. /// Read by the dashboard to render a spinner; cleared when the action /// resolves (success or failure). - transient: Mutex>, + /// Transients for work with NO queue node behind it (destroy, migration). + /// Queue-driven pills are derived, not stored — see `transient_snapshot`. + manual_transient: Mutex>, /// Tombstone for transients that have JUST been cleared. The /// crash watcher polls every 10s and would race the /// drop-clears-immediately path of `TransientGuard`: an operator @@ -381,7 +384,10 @@ pub struct TransientState { /// 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, + /// 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, } /// RAII handle returned by `Coordinator::transient_guard`. Cleared on @@ -398,7 +404,7 @@ pub struct TransientGuard { impl Drop for TransientGuard { fn drop(&mut self) { - self.coord.clear_transient(&self.name); + self.coord.clear_manual_transient(&self.name); } } @@ -537,7 +543,7 @@ impl Coordinator { agent_io_weight, model_prices, agents: Mutex::new(HashMap::new()), - transient: Mutex::new(HashMap::new()), + manual_transient: Mutex::new(HashMap::new()), recent_transient: Mutex::new(HashMap::new()), recent_crashes: Mutex::new(HashMap::new()), graceful_stop_pending: Mutex::new(HashSet::new()), @@ -1077,25 +1083,31 @@ impl Coordinator { /// Mark an agent as in-progress (only one state per agent for now). /// - /// Two callers, for two different reasons: - /// - **Work with a queue node behind it** — the job-queue scheduler, which - /// publishes the edges of a *derived* set - /// ([`crate::job_queue::JobQueue::running_transients`]). It needs no guard: - /// nothing is owned, and a node that stops running stops appearing. - /// - **Work with no node** (destroy, migration) — via the RAII - /// [`TransientGuard`], so the paired `clear_transient` still runs if the - /// surrounding future is cancelled (HTTP request aborted, shutdown - /// mid-rebuild, panic). There a bare set really would leak the transient - /// and pin the dashboard on "rebuilding…" forever. - pub(crate) fn set_transient(&self, name: &str, label: String, deliberate_stop: bool) { - self.transient.lock().unwrap().insert( + /// Record a transient for work that has **no queue node behind it** — + /// destroy and migration. Reached only through the RAII [`TransientGuard`], + /// so the paired clear still runs if the surrounding future is cancelled + /// (HTTP request aborted, shutdown mid-rebuild, panic); a bare set here + /// really would leak the transient and pin the dashboard on "rebuilding…". + /// + /// Queue-driven work does **not** come through here. Its pills are derived + /// from the running graph ([`crate::job_queue::JobQueue::running_transients`]) + /// and merged in by [`Coordinator::transient_snapshot`] — nothing is stored, + /// so nothing can go stale or leak. + fn set_manual_transient(&self, name: &str, label: String, deliberate_stop: bool) { + self.manual_transient.lock().unwrap().insert( name.to_owned(), TransientState { label: label.clone(), deliberate_stop, - since: std::time::Instant::now(), + since: hive_sh4re::wire_time::from_secs(hive_sh4re::wire_time::now_unix()), }, ); + self.emit_transient_set(name, label); + } + + /// Emit the "a pill appeared" edge. Shared by the manual path and the + /// job-queue scheduler, which publishes the transitions of its derived set. + pub(crate) fn emit_transient_set(&self, name: &str, label: String) { // Live-update dashboards. `since_unix` is wall-clock so the // browser can tick "Ns spawning…" without polling. The // intra-process map keeps using `Instant` for monotonicity. @@ -1112,30 +1124,37 @@ impl Coordinator { }); } - /// Clear an agent's transient state. Reached either from - /// [`TransientGuard`]'s `Drop` (the no-node callers) or from the scheduler - /// when a derived pill stops being current — see - /// [`Coordinator::set_transient`]. - pub(crate) fn clear_transient(&self, name: &str) { - let removed = self.transient.lock().unwrap().remove(name); + /// Drop a manual transient (see [`Coordinator::set_manual_transient`]). + /// Reached only from [`TransientGuard`]'s `Drop`, which guarantees it runs. + fn clear_manual_transient(&self, name: &str) { + let removed = self.manual_transient.lock().unwrap().remove(name); if let Some(state) = removed { - // Stamp the tombstone so the crash watcher can still see - // "operator kicked this off recently" on its next 10s poll - // — without this, the clear-then-poll race produced a - // 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.deliberate_stop, std::time::Instant::now()), - ); - self.emit_dashboard_event(DashboardEvent::TransientCleared { - seq: self.next_seq(), - name: name.to_owned(), - }); + self.emit_transient_cleared(name, state.deliberate_stop); } } + /// Emit the "a pill went away" edge **and stamp the tombstone the crash + /// watcher reads**. Shared by the manual path and the job-queue scheduler. + /// + /// 🚨 The stamp is not bookkeeping. Without it the clear-then-poll race + /// produced a spurious `ContainerCrash` on **every** operator stop/restart: + /// the transient is gone by the time the 10s poll looks, so a deliberate + /// stop is indistinguishable from a crash. `recent_transient_within` is what + /// closes that window, which is why the clear has to be an *event* — a + /// 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) { + self.recent_transient.lock().unwrap().insert( + name.to_owned(), + (deliberate_stop, std::time::Instant::now()), + ); + self.emit_dashboard_event(DashboardEvent::TransientCleared { + seq: self.next_seq(), + name: name.to_owned(), + }); + } + /// Mark `name` as having a graceful stop in progress. While set, /// `socket_server::handle_recv` returns `Response::GracefulStop` for /// this agent instead of polling the broker (the inbound fence). @@ -1259,15 +1278,46 @@ impl Coordinator { label: impl Into, deliberate_stop: bool, ) -> TransientGuard { - self.set_transient(name, label.into(), deliberate_stop); + self.set_manual_transient(name, label.into(), deliberate_stop); TransientGuard { coord: self.clone(), name: name.to_owned(), } } + /// Every live transient, keyed by agent. + /// + /// **Derived on read**, not stored: the queue-driven pills come straight + /// from the running graph, so there is no cached copy to go stale, leak, or + /// disagree with what is actually running. The only stored entries are the + /// handful with no node behind them (destroy, migration), overlaid on top — + /// they win, since an agent being destroyed is the more urgent truth than + /// whatever node was mid-flight when it started. + #[must_use] pub fn transient_snapshot(&self) -> HashMap { - self.transient.lock().unwrap().clone() + let mut out: 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(); + out.extend( + self.manual_transient + .lock() + .unwrap() + .iter() + .map(|(k, v)| (k.clone(), v.clone())), + ); + 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 0016487a..04c43935 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -535,7 +535,13 @@ fn build_transient_views( .map(|(name, st)| TransientView { name: name.clone(), kind: st.label.clone(), - secs: st.since.elapsed().as_secs(), + // Clamped at 0: `since` is wall-clock now (the node's own + // `started_at`), so a backwards clock adjustment could otherwise + // render a negative age. + secs: (hive_sh4re::wire_time::from_secs(hive_sh4re::wire_time::now_unix()) - st.since) + .num_seconds() + .max(0) + .cast_unsigned(), }) .collect() } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 71f6263e..ada7faad 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -59,6 +59,27 @@ const MAX_HISTORY_DAGS: usize = 50; /// Cap on stored node error strings. const MAX_ERROR_LEN: usize = 2_000; +/// One live transient pill, derived from a running node. +/// +/// A named struct rather than a tuple because three of its four fields are +/// easy to confuse at a call site: two are strings and two answer questions +/// nobody should have to guess at ("is this the agent or the label?", "does +/// this bool mean deliberate or running?"). +#[derive(Debug, Clone)] +pub struct RunningTransient { + /// The agent whose lease the node declared. + pub agent: String, + /// The node's own wire tag, rendered as the pill. + pub label: String, + /// Whether this operation is expected to take the container down — the + /// crash watcher's input. See [`NodeKind::takes_container_down`]. + pub takes_container_down: bool, + /// When the node started running, so the dashboard can tick elapsed + /// seconds. Taken from the node itself, which is the true start of the + /// operation rather than the moment a watcher noticed it. + pub since: DateTime, +} + /// A node claimed for execution — everything the executor needs, snapshotted at /// claim time. #[derive(Debug, Clone)] @@ -422,7 +443,7 @@ impl JobQueue { /// the resources-where-constructed work, not this function. An agent's lease /// is cap-1, so at most one entry per agent. #[must_use] - pub fn running_transients(&self) -> Vec<(String, String, bool)> { + pub fn running_transients(&self) -> Vec { let inner = self.lock(); inner .sched @@ -441,11 +462,17 @@ impl JobQueue { } => Some(a), _ => None, })?; - Some(( + Some(RunningTransient { agent, - n.payload.as_str().to_owned(), - n.payload.takes_container_down(), - )) + 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 + // only sees `Running` nodes — the fallback is unreachable in + // practice, and "just now" is the honest answer if it isn't. + since: n + .started_at + .unwrap_or_else(|| hive_sh4re::wire_time::from_secs(now_unix())), + }) }) .collect() } diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index 492aa55d..105d5326 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -59,7 +59,7 @@ pub async fn run_worker(coord: Arc) { // 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(); + 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 @@ -174,25 +174,29 @@ fn handle_completion(coord: &Arc, done: NodeDone) { /// 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 HashMap) { 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. - prev.retain(|agent, label| { - let still = running.iter().any(|(a, l, _)| a == agent && l == label); + // 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)| { + let still = running + .iter() + .any(|t| &t.agent == agent && &t.label == label); if !still { - coord.clear_transient(agent); + coord.emit_transient_cleared(agent, *deliberate); } still }); - for (agent, label, takes_down) in running { - if prev.get(&agent) == Some(&label) { + for t in running { + if prev.get(&t.agent).map(|(l, _)| l) == Some(&t.label) { continue; } - coord.set_transient(&agent, label.clone(), takes_down); - prev.insert(agent, label); + coord.emit_transient_set(&t.agent, t.label.clone()); + prev.insert(t.agent, (t.label, t.takes_container_down)); } }