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

refactor(#2815): the scheduler publishes pill edges, it doesn't own pills

Browse source 
mara on !2910: "transient guard as well - should be removable now?" — for
the queue path, yes.

`set_transient`'s own doc explained why the RAII guard existed: a
cancelled future must not leak an imperatively-set transient and pin the
dashboard on "rebuilding…" forever. That cannot happen to a derived set.
`running_transients()` is recomputed from the graph every loop, so a node
that stops running stops appearing — there is nothing to own and nothing
to leak.

So the scheduler no longer holds a guard per pill. It keeps the previous
derived value and publishes the transitions, which is the one thing a
derived read cannot express: the dashboard wants `TransientSet` /
`TransientCleared` edges, and the crash watcher wants the *moment* a pill
cleared, since its grace window is what stops an operator stop from
reading as a crash.

That also retires a hazard rather than restating it. The old code carried
a warning that stale guards had to be dropped before new ones were
created, because `TransientGuard::drop` clears by agent with no notion of
which label it was clearing — so a same-agent label change could clear
the pill it had just set. With no guards there is no ordering to get
wrong; clears are emitted before sets so a relabel reads as
clear-then-set rather than two overlapping pills.

`set_transient` / `clear_transient` become `pub(crate)`. The guard stays
for destroy and migration, which have no node behind them and where the
cancellation concern is real.

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 17:30:05 +02:00
commit 56202065d5
2 changed files with 57 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

66
hive-c0re/src/job_queue/scheduler.rs
View file

@ -54,10 +54,12 @@ struct NodeDone {
pub async fn run_worker(coord: Arc<Coordinator>) {
let mut shutdown = coord.shutdown_rx();
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<NodeDone>();
// 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<String, (String, crate::coordinator::TransientGuard)> =
HashMap::new();
// Last derived pill set we published, keyed by agent (its lease is cap-1,
// so one pill each). Purely the previous value of a *derived* quantity —
// 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<String, String> = HashMap::new();
loop {
// Checked every iteration, not just in the `select!` below — a
// continuous stream of ready claims never reaches the `select!`, so
@ -148,10 +150,23 @@ fn handle_completion(coord: &Arc<Coordinator>, done: NodeDone) {
coord.emit_rebuild_queue_snapshot();
}
/// 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.
/// Publish the transitions between the previously-derived pill set and the
/// current one. `prev` is last loop's derived value, keyed by agent (an agent's
/// lease is cap-1, so at most one pill each).
///
/// The pill set itself isn't owned or stored here — it is
/// [`super::JobQueue::running_transients`], recomputed from the graph. What this
/// publishes is the **edges**, which a derived read can't express on its own:
/// the dashboard wants `TransientSet` / `TransientCleared` events, and the crash
/// watcher wants the *moment* a pill cleared (its grace window is what keeps an
/// operator stop from reading as a crash).
///
/// There used to be an RAII `TransientGuard` per pill here, and a hazard note
/// about dropping stale guards before creating new ones or a same-agent label
/// change would clear the pill it had just set. Both are gone: a guard exists so
/// a cancelled future can't *leak* an imperatively-set transient, and a derived
/// set has nothing to leak — a node that stops running simply stops appearing.
/// (Destroy and migration still take guards; they have no node behind them.)
///
/// `deliberate_stop` rides along per node ([`NodeKind::takes_container_down`])
/// rather than being blanket-`true` for anything holding a lease: `Create` and
@ -159,24 +174,25 @@ fn handle_completion(coord: &Arc<Coordinator>, done: NodeDone) {
/// 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<Coordinator>,
transients: &mut HashMap<String, (String, crate::coordinator::TransientGuard)>,
) {
fn reconcile_transients(coord: &Arc<Coordinator>, prev: &mut HashMap<String, String>) {
let running = coord.job_queue.running_transients();
transients.retain(|agent, (label, _)| running.iter().any(|(a, l, _)| a == agent && l == label));
// 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);
if !still {
coord.clear_transient(agent);
}
still
});
for (agent, label, takes_down) in running {
transients.entry(agent.clone()).or_insert_with(|| {
let guard = coord.transient_guard(&agent, label.clone(), takes_down);
(label, guard)
});
if prev.get(&agent) == Some(&label) {
continue;
}
coord.set_transient(&agent, label.clone(), takes_down);
prev.insert(agent, label);
}
}