diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index b2a354ea..d3eb90c4 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -882,9 +882,10 @@ 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. - // 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); + // Destroy has no queue node behind it, so nothing in the graph says this + // container is going away on purpose — without this the crash watcher + // reports every destroy as a crash and the manager tries to recover it. + let guard = coord.suppress_crash_watch(name); 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 ade13914..d03ec18a 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -119,9 +119,10 @@ 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). - /// Transients for work with NO queue node behind it (destroy, migration). - /// Queue-driven pills are derived, not stored — see `transient_snapshot`. - manual_transient: Mutex>, + /// Agents whose container is being taken down by work with **no queue node + /// behind it** (destroy, migration), so the crash watcher must not report + /// the disappearance as a crash. Not a pill — see [`CrashWatchSuppression`]. + crash_suppressed: 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 @@ -390,21 +391,45 @@ pub struct TransientState { pub since: DateTime, } -/// RAII handle returned by `Coordinator::transient_guard`. Cleared on -/// drop — including drop-via-cancellation, the path that bare -/// `set_transient` / `clear_transient` pairs leaked through. Holds an -/// `Arc` so the guard is freely returnable / movable. -#[must_use = "the guard clears the transient when dropped; bind it for the operation's \ - duration (`let _guard = coord.transient_guard(...)`). An unbound call drops \ - it immediately and un-sets the transient at once — the exact footgun this guards against."] -pub struct TransientGuard { +/// RAII handle returned by [`Coordinator::suppress_crash_watch`]. While held, +/// the crash watcher treats this container disappearing as **expected**. +/// +/// This is *not* a dashboard pill. Transients are derived from running queue +/// nodes and nothing stores them. But destroy and migration take a container +/// down without a node behind them, so nothing in the graph says the +/// disappearance was intended — and without that, `crash_watch` fires a +/// `ContainerCrash` for every destroy and every migrated agent, and the manager +/// tries to "recover" containers that were removed on purpose. +/// +/// It is held rather than stamped once because +/// [`crate::workers::crash_watch`]'s grace window is finite and these +/// operations are not: a long destroy would outlive a single tombstone. The +/// tombstone is stamped on drop, covering the poll that lands just after. +/// +/// Goes away entirely once destroy + migration are real queue nodes. +#[must_use = "suppression lasts as long as the guard; bind it for the operation's duration \ + (`let _guard = coord.suppress_crash_watch(...)`). An unbound call drops it \ + immediately and the very next poll can report a deliberate stop as a crash."] +pub struct CrashWatchSuppression { coord: Arc, name: String, } -impl Drop for TransientGuard { +impl Drop for CrashWatchSuppression { fn drop(&mut self) { - self.coord.clear_manual_transient(&self.name); + self.coord + .crash_suppressed + .lock() + .unwrap() + .remove(&self.name); + // 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())); } } @@ -543,7 +568,7 @@ impl Coordinator { agent_io_weight, model_prices, agents: Mutex::new(HashMap::new()), - manual_transient: Mutex::new(HashMap::new()), + crash_suppressed: Mutex::new(HashSet::new()), recent_transient: Mutex::new(HashMap::new()), recent_crashes: Mutex::new(HashMap::new()), graceful_stop_pending: Mutex::new(HashSet::new()), @@ -1081,32 +1106,8 @@ impl Coordinator { self.agents.lock().unwrap().keys().cloned().collect() } - /// Mark an agent as in-progress (only one state per agent for now). - /// - /// 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: 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. + /// Emit the "a pill appeared" edge, for the job-queue scheduler publishing + /// 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 @@ -1124,17 +1125,8 @@ impl Coordinator { }); } - /// 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 { - 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. + /// watcher reads**, for 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: @@ -1272,31 +1264,37 @@ impl Coordinator { /// 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_manual_transient(name, label.into(), deliberate_stop); - TransientGuard { + pub fn suppress_crash_watch(self: &Arc, name: &str) -> CrashWatchSuppression { + self.crash_suppressed + .lock() + .unwrap() + .insert(name.to_owned()); + CrashWatchSuppression { coord: self.clone(), name: name.to_owned(), } } + /// Whether a no-node operation is currently taking this container down. + #[must_use] + pub fn crash_watch_suppressed(&self, name: &str) -> bool { + self.crash_suppressed.lock().unwrap().contains(name) + } + /// 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. + /// **Derived on read, stored nowhere.** Straight off the running graph, so + /// there is no cached copy to go stale, leak, or disagree with what is + /// actually running. + /// + /// Work with no queue node behind it (destroy, migration) therefore shows + /// **no pill** — there is nothing in the graph to derive one from. Its + /// crash-watch suppression is a separate, narrower thing + /// ([`Coordinator::suppress_crash_watch`]); the pill comes back for free + /// once those become real nodes. #[must_use] pub fn transient_snapshot(&self) -> HashMap { - let mut out: HashMap = self - .job_queue + self.job_queue .running_transients() .into_iter() .map(|t| { @@ -1309,15 +1307,7 @@ impl Coordinator { }, ) }) - .collect(); - out.extend( - self.manual_transient - .lock() - .unwrap() - .iter() - .map(|(k, v)| (k.clone(), v.clone())), - ); - out + .collect() } /// Drop a system message into the given agent's inbox. Wakes the diff --git a/hive-c0re/src/migrate.rs b/hive-c0re/src/migrate.rs index 38a5d5e2..56f1f0bb 100644 --- a/hive-c0re/src/migrate.rs +++ b/hive-c0re/src/migrate.rs @@ -109,9 +109,8 @@ pub async fn run(coord: &Arc) -> Result<()> { // would fire ContainerCrash for every agent here and the // manager would spuriously try to recover them. // 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); + // directly — so nothing in the graph marks the stop as intended. + let guard = coord.suppress_crash_watch(name.as_str()); let result = repoint_container(name.as_str()).await; drop(guard); if let Err(e) = result { @@ -203,8 +202,8 @@ async fn rename_manager_container(coord: &Arc) { return; } tracing::info!("migration phase 5: renaming root container to h-root"); - // `true`: the old container is stopped immediately below. - let _guard = coord.transient_guard(MANAGER_NAME, "rebuilding", true); + // The old container is stopped immediately below, on purpose. + let _guard = coord.suppress_crash_watch(MANAGER_NAME); // 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/crash_watch.rs b/hive-c0re/src/workers/crash_watch.rs index 9b74f826..73999f1c 100644 --- a/hive-c0re/src/workers/crash_watch.rs +++ b/hive-c0re/src/workers/crash_watch.rs @@ -87,7 +87,13 @@ 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.deliberate_stop); + // 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. + let active = transients + .get(stopped) + .map(|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) { continue;