refactor(#2441): update queue consumers + docs for agent-per-node

DagView no longer has a DAG-level agent, so consumers derive it from the
per-node agents:
- hivectl dag_progress.rs: dag_agents(d) helper (distinct node agents,
  comma-joined) in place of d.agent.
- dashboard builds.js: entryAgents(entry) helper likewise for the
  rebuild-queue card + live-log header + cancel confirms.
- docs/coordinator.md: lease prose (node-agent-keyed, global per agent),
  wire shape (NodeView.agent, no DagView.agent), and dropped the removed
  dedup section.
This commit is contained in:
atlas 2026-07-14 21:42:37 +02:00 committed by mara
commit e2b48d2014
3 changed files with 66 additions and 36 deletions

View file

@ -22,7 +22,7 @@ The **DAG** is the unit of dedup / cancel / approval-resolution and the
dashboard group; the **node** is the unit of scheduling / execution / dashboard group; the **node** is the unit of scheduling / execution /
build-log / step label. Deps are intra-DAG edges only (`AfterOk` by default: build-log / step label. Deps are intra-DAG edges only (`AfterOk` by default:
the dep must succeed, a failed/cancelled dep cancels the dependent — the dep must succeed, a failed/cancelled dep cancels the dependent —
cancel-downstream). Cross-DAG ordering comes from the per-agent lease + dedup, cancel-downstream). Cross-DAG ordering comes from the per-agent lease,
never from edges between DAGs. Submit-time validation (petgraph `toposort`) never from edges between DAGs. Submit-time validation (petgraph `toposort`)
rejects cyclic specs outright, fixing the old queue's "circular dep silently rejects cyclic specs outright, fixing the old queue's "circular dep silently
deadlocks" caveat. deadlocks" caveat.
@ -139,11 +139,13 @@ resources are free. Resources:
1. **Build slots**`services.hyperhive.c0re.buildSlots` permits (default 1), 1. **Build slots**`services.hyperhive.c0re.buildSlots` permits (default 1),
held by nix-heavy nodes for the node's duration. held by nix-heavy nodes for the node's duration.
2. **Per-agent lifecycle lease** — DAG-scoped: acquired at the DAG's first 2. **Per-agent lifecycle lease** — keyed on the **node's** agent (agent is
container-affecting node (`StopForUpdate`, `Swap`, `Signal`, `Drain`, per-node; a DAG can span agents) and globally exclusive per agent across
`Reconcile`, `WriteDropin`, `Create`, `ApprovalDeploy`), held until the DAG all DAGs: acquired at a container-affecting node (`StopForUpdate`, `Swap`,
is terminal, so two lifecycle DAGs for one agent never interleave their `Signal`, `Drain`, `Reconcile`, `WriteDropin`, `Create`, `ApprovalDeploy`),
container ops. **Lease-exempt**: `Prebuild`, `MetaLock`, `WritePermFile` held by the owning DAG until it's terminal, so two DAGs never interleave
container ops on the same agent. A DAG touching several agents holds one
lease per agent. **Lease-exempt**: `Prebuild`, `MetaLock`, `WritePermFile`
they touch the store / meta, not the running container, which is exactly they touch the store / meta, not the running container, which is exactly
why a stop can land while another DAG's prebuild is still building. why a stop can land while another DAG's prebuild is still building.
@ -156,14 +158,11 @@ The queue is in-memory only and lost on hive-c0re restart — deliberate:
desired state is re-derived at boot from the DB + rev markers (see _Boot desired state is re-derived at boot from the DB + rev markers (see _Boot
reconcile_), so there is no durable-recovery machinery to go wrong. reconcile_), so there is no durable-recovery machinery to go wrong.
### Dedup, cancel, history ### Cancel, history
Dedup at **DAG granularity**: a repeat submit against a DAG whose roll-up is Submit-time dedup was removed with the agent-per-node move (a multi-agent DAG
still `Queued` with the same `(template, agent, parent_id, approval_id)` has no single agent to key a dedup on), so every submit enqueues a fresh DAG;
plus `inputs` for meta-updates and the perm-type discriminant for perm whether any dedup needs reintroducing is tracked as a follow-up.
changes — returns the existing id and appends an "also requested by …" line.
`parent_id` in the key keeps a cascade child from collapsing into a
standalone or sweep rebuild. Running/terminal DAGs never dedup.
Cancel only applies to still-fully-queued DAGs (an in-flight nix build isn't Cancel only applies to still-fully-queued DAGs (an in-flight nix build isn't
interruptible); `cancel_children` cancels a parent's still-queued child DAGs. interruptible); `cancel_children` cancels a parent's still-queued child DAGs.
@ -186,12 +185,14 @@ of dangling it).
### Wire shape ### Wire shape
`RebuildQueueChanged { seq, queue: [DagView…] }` (event name kept). Each `RebuildQueueChanged { seq, queue: [DagView…] }` (event name kept). Each
`DagView` carries the old entry-level fields (`id`, `kind` = template string, `DagView` carries the entry-level fields (`id`, `kind` = template string,
roll-up `state`, `agent`, `source`, `parent_id`, `reason`, timestamps, roll-up `state`, `source`, `parent_id`, `reason`, timestamps, `inputs`,
`inputs`, `approval_id`) plus `nodes: [NodeView…]` — per-node `kind`, `deps`, `approval_id`) plus `nodes: [NodeView…]` — per-node `agent`, `kind`, `deps`,
`state`, `step`, `build_log_id`, timestamps, `error`. Step labels and build `state`, `step`, `build_log_id`, timestamps, `error`. There is **no
logs are **per-node**; the dashboard renders the node chain on each queue DAG-level `agent`** (agent is per-node, so a DAG can span agents); consumers
card and keys the live-log panel off the running node. derive a DAG's agent(s) from its nodes. Step labels and build logs are
**per-node**; the dashboard renders the node chain on each queue card and
keys the live-log panel off the running node.
--- ---

View file

@ -165,11 +165,22 @@ function firstFailedNode(entry) {
return (entry.nodes || []).find((n) => n.state === 'failed') || null; return (entry.nodes || []).find((n) => n.state === 'failed') || null;
} }
// Distinct agents across a DAG's nodes, comma-joined for display. Agent is
// per-node now (a DAG can span agents, e.g. a hive-wide restart), so there's
// no DAG-level `agent` field — derive it from the nodes.
function entryAgents(entry) {
const seen = [];
for (const n of entry.nodes || []) {
if (n.agent && !seen.includes(n.agent)) seen.push(n.agent);
}
return seen.join(',');
}
function rebuildQueueEntryFingerprint(entry, isChild) { function rebuildQueueEntryFingerprint(entry, isChild) {
return JSON.stringify({ return JSON.stringify({
state: entry.state, state: entry.state,
kind: entry.kind, kind: entry.kind,
agent: entry.agent, agent: entryAgents(entry),
source: entry.source, source: entry.source,
started_at: entry.started_at, started_at: entry.started_at,
enqueued_at: entry.enqueued_at, enqueued_at: entry.enqueued_at,
@ -265,7 +276,7 @@ function renderQueueEntry(entry, _byId, isChild) {
el('span', { class: 'rqe-kind', title: entry.kind }, el('span', { class: 'rqe-kind', title: entry.kind },
(QUEUE_KIND_GLYPH[entry.kind] || '?') + ' ' + entry.kind), (QUEUE_KIND_GLYPH[entry.kind] || '?') + ' ' + entry.kind),
' ', ' ',
el('code', { class: 'rqe-agent' }, entry.agent), el('code', { class: 'rqe-agent' }, entryAgents(entry)),
); );
li.append(' ', el('span', { class: 'rqe-source rqe-source-' + entry.source }, entry.source)); li.append(' ', el('span', { class: 'rqe-source rqe-source-' + entry.source }, entry.source));
if (entry.state === 'queued') { if (entry.state === 'queued') {
@ -331,14 +342,14 @@ function renderQueueEntry(entry, _byId, isChild) {
class: 'inline rqe-cancel', class: 'inline rqe-cancel',
'data-async': '', 'data-async': '',
'data-confirm': 'data-confirm':
`cancel ${entry.kind} for \`${entry.agent}\` (queue id ${entry.id})? ` + `cancel ${entry.kind} for \`${entryAgents(entry)}\` (queue id ${entry.id})? ` +
`the row drops from the queue and never runs. running / done / failed entries can't be cancelled this way.`, `the row drops from the queue and never runs. running / done / failed entries can't be cancelled this way.`,
}); });
cancelForm.append(el('button', { cancelForm.append(el('button', {
type: 'submit', type: 'submit',
class: 'rqe-cancel-btn', class: 'rqe-cancel-btn',
title: 'cancel this queued ' + entry.kind, title: 'cancel this queued ' + entry.kind,
'aria-label': 'cancel queued ' + entry.kind + ' for ' + entry.agent, 'aria-label': 'cancel queued ' + entry.kind + ' for ' + entryAgents(entry),
}, '✗')); }, '✗'));
li.append(cancelForm); li.append(cancelForm);
} }
@ -414,7 +425,7 @@ function renderRebuildLiveLog(queue) {
const header = el('div', { class: 'rebuild-live-log-header' }, const header = el('div', { class: 'rebuild-live-log-header' },
toggle, ' ', toggle, ' ',
el('span', { class: 'rebuild-live-log-title' }, 'live build log — '), el('span', { class: 'rebuild-live-log-title' }, 'live build log — '),
el('code', { class: 'rqe-agent' }, running.agent), el('code', { class: 'rqe-agent' }, entryAgents(running)),
' ', el('span', { class: 'rqe-kind' }, ' ', el('span', { class: 'rqe-kind' },
(running.kind || 'rebuild') + ' · ' + (NODE_KIND_LABEL[liveNode.kind] || liveNode.kind)), (running.kind || 'rebuild') + ' · ' + (NODE_KIND_LABEL[liveNode.kind] || liveNode.kind)),
' ', badge, ' ', ' ', badge, ' ',

View file

@ -68,7 +68,7 @@ async fn wait_for_dags_plain(socket: &Path, ids: Vec<u64>) -> Result<()> {
// sees whether the agent came back. // sees whether the agent came back.
if d.nodes.iter().all(|n| n.state.is_terminal()) { if d.nodes.iter().all(|n| n.state.is_terminal()) {
if d.state == hive_sh4re::jobs::State::Failed { if d.state == hive_sh4re::jobs::State::Failed {
failed.push(format!("{} {}", d.kind.as_str(), d.agent)); failed.push(format!("{} {}", d.kind.as_str(), dag_agents(d)));
} }
} else { } else {
all_terminal = false; all_terminal = false;
@ -138,7 +138,7 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec<u64>) -> Result<()> {
"{} {} {} · {}", "{} {} {} · {}",
state_glyph(d.state), state_glyph(d.state),
d.kind.as_str(), d.kind.as_str(),
d.agent, dag_agents(d),
fmt_dur(dag_elapsed(d, now)), fmt_dur(dag_elapsed(d, now)),
)); ));
for n in &d.nodes { for n in &d.nodes {
@ -166,7 +166,7 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec<u64>) -> Result<()> {
} }
if d.nodes.iter().all(|n| n.state.is_terminal()) { if d.nodes.iter().all(|n| n.state.is_terminal()) {
if d.state == hive_sh4re::jobs::State::Failed { if d.state == hive_sh4re::jobs::State::Failed {
failed.push(format!("{} {}", d.kind.as_str(), d.agent)); failed.push(format!("{} {}", d.kind.as_str(), dag_agents(d)));
} }
} else { } else {
all_terminal = false; all_terminal = false;
@ -204,6 +204,19 @@ fn finish_wait(mut failed: Vec<String>) -> Result<()> {
} }
} }
/// Distinct agents across a DAG's nodes, comma-joined for display — the
/// per-node replacement for the old DAG-level `agent` field. Single-agent
/// DAGs render one name; a hive-wide DAG lists each.
fn dag_agents(d: &hive_sh4re::jobs::DagView) -> String {
let mut seen: Vec<&str> = Vec::new();
for n in &d.nodes {
if !seen.contains(&n.agent.as_str()) {
seen.push(&n.agent);
}
}
seen.join(",")
}
/// Current unix time in seconds (0 on the impossible pre-epoch error). /// Current unix time in seconds (0 on the impossible pre-epoch error).
fn now_unix() -> i64 { fn now_unix() -> i64 {
std::time::SystemTime::now() std::time::SystemTime::now()
@ -290,7 +303,7 @@ fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String {
"{} {} {:<12}", "{} {} {:<12}",
state_glyph(d.state), state_glyph(d.state),
d.kind.as_str(), d.kind.as_str(),
d.agent dag_agents(d)
); );
for (i, n) in d.nodes.iter().enumerate() { for (i, n) in d.nodes.iter().enumerate() {
let sep = if i == 0 { " " } else { "" }; let sep = if i == 0 { " " } else { "" };
@ -314,9 +327,10 @@ mod tests {
use super::render_dag_line; use super::render_dag_line;
fn node(id: u32, kind: &str, state: State, step: Option<&str>) -> NodeView { fn node(id: u32, agent: &str, kind: &str, state: State, step: Option<&str>) -> NodeView {
NodeView { NodeView {
id, id,
agent: agent.to_owned(),
kind: kind.to_owned(), kind: kind.to_owned(),
deps: if id == 0 { vec![] } else { vec![id - 1] }, deps: if id == 0 { vec![] } else { vec![id - 1] },
state, state,
@ -332,7 +346,6 @@ mod tests {
fn render_dag_line_shows_chain_and_running_step() { fn render_dag_line_shows_chain_and_running_step() {
let dag = DagView { let dag = DagView {
id: 7, id: 7,
agent: "alice".to_owned(),
kind: Template::Rebuild, kind: Template::Rebuild,
state: State::Running, state: State::Running,
source: Source::Manual, source: Source::Manual,
@ -345,10 +358,16 @@ mod tests {
approval_id: None, approval_id: None,
perm_payload: None, perm_payload: None,
nodes: vec![ nodes: vec![
node(0, "prebuild", State::Done, None), node(0, "alice", "prebuild", State::Done, None),
node(1, "stop_for_update", State::Done, None), node(1, "alice", "stop_for_update", State::Done, None),
node(2, "swap", State::Running, Some("nixos-container update")), node(
node(3, "reconcile", State::Queued, None), 2,
"alice",
"swap",
State::Running,
Some("nixos-container update"),
),
node(3, "alice", "reconcile", State::Queued, None),
], ],
}; };
let line = render_dag_line(&dag); let line = render_dag_line(&dag);
@ -363,11 +382,10 @@ mod tests {
#[test] #[test]
fn render_dag_line_surfaces_first_node_error() { fn render_dag_line_surfaces_first_node_error() {
let mut failed = node(0, "prebuild", State::Failed, None); let mut failed = node(0, "bob", "prebuild", State::Failed, None);
failed.error = Some("nix build exploded".to_owned()); failed.error = Some("nix build exploded".to_owned());
let dag = DagView { let dag = DagView {
id: 8, id: 8,
agent: "bob".to_owned(),
kind: Template::Rebuild, kind: Template::Rebuild,
state: State::Failed, state: State::Failed,
source: Source::Manual, source: Source::Manual,