vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(#2815): drop the imperative transient path entirely

Browse source 
mara on !2910: "also remove imperative path for the things that are not
nodes yet, file follow up issue to fix that".

`TransientGuard`, the stored map and both manual set/clear are gone.
`transient_snapshot()` is derived and nothing else — destroy and
migration show no pill, because there is no node to derive one from. The
pill returns for free when they become nodes.

What those three guards were actually doing, though, was suppressing the
crash watcher, not drawing a pill. `migrate.rs` said so in its own
comment: without it, `crash_watch` fires `ContainerCrash` for every
migrated agent and the manager tries to recover containers that were
stopped on purpose. Destroy is the same — the container disappears
deliberately and nothing in the graph says so.

Deleting them outright would therefore have traded a dashboard pill for
false crash alerts on every destroy and every migration. So the
suppression survives as its own thing, `suppress_crash_watch`, with a
name that says what it is. It is still RAII, and still held for the
operation rather than stamped once, because the crash watcher's grace
window is finite and a destroy is not — a single tombstone would expire
mid-operation. The drop stamps the tombstone, covering the poll that
lands just after.

That leaves RAII in the codebase for exactly one purpose instead of two.
Untangling the pill from the suppression is what made the transient layer
deletable at all.

Follow-up issue for making destroy + migration real queue nodes to
follow; at that point this guard goes too.

Checked with clippy (`--all-targets -D warnings`), `cargo test -p
hive-c0re -p hive-jobq` (322 + 41 passed) and `nix fmt`.
This commit is contained in:
atlas 2026-08-01 18:15:27 +02:00
commit 7f920718f2
4 changed files with 80 additions and 84 deletions
Download patch file Download diff file Expand all files Collapse all files

7
hive-c0re/src/actions.rs
View file

@ -882,9 +882,10 @@ pub async fn destroy(coord: &Arc<Coordinator>, 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);

140
hive-c0re/src/coordinator.rs
View file

@ -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<HashMap<String, TransientState>>,
/// 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<HashSet<String>>,
/// 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<Utc>,
}
/// 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<Coordinator>` 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<Coordinator>,
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<Self>,
name: &str,
label: impl Into<String>,
deliberate_stop: bool,
) -> TransientGuard {
self.set_manual_transient(name, label.into(), deliberate_stop);
TransientGuard {
pub fn suppress_crash_watch(self: &Arc<Self>, 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<String, TransientState> {
let mut out: HashMap<String, TransientState> = 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

9
hive-c0re/src/migrate.rs
View file

@ -109,9 +109,8 @@ pub async fn run(coord: &Arc<Coordinator>) -> 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<Coordinator>) {
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

8
hive-c0re/src/workers/crash_watch.rs
View file

@ -87,7 +87,13 @@ fn emit_crash_transitions(coord: &Coordinator, prev: &HashSet<String>, 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;