job_queue: retire the now-off-wire step sub-step label
The `step` label was taken off the wire in #2661, when each deploy phase became a first-class DAG node. Since then it has been written but never read: `NodeRuntime` derives only `Debug, Default, Clone` — no serde — so the field could not reach any client, and the only reads of it were the dedup checks inside its own setters. This deletes the machinery. Removed: - `NodeRuntime.step`, `set_step`, `set_step_running`, and the `rt.step = None` clear in `complete_node`. `NodeRuntime` keeps its remaining `build_log_id` field (deliberately still a struct — collapsing it to a bare `Option<i64>` would churn every call site for no gain). - `Ctx::step` and its ~15 call sites in `job_queue/exec.rs`. `Ctx` itself stays: it is the build-log sink, which `run_prebuild` and `run_swap` still use. - `Coordinator::set_queue_step` and its 11 callers in `actions.rs`. - `JobQueue::running_node_of`, reachable only from `set_queue_step`. - `swap_update`'s `on_step` parameter and its one body call. - The `set_step_only_on_running_and_signals_change` test. Dropping the calls orphaned parameters, which are removed with their call sites: `ctx` on ten executors that used it only as a step sink, and `queue_entry_id` on `run_deploy_merge_verify` / `run_deploy_apply` / `run_finalize_deploy` plus both `coord` and `queue_entry_id` on `prepare_applied_target`. `run_deploy_tail` KEEPS its `queue_entry_id` — that one has a genuine surviving use (the build-log link in the failure comment posted to the PR). One behavioural change, called out so it is not mistaken for a dropped dashboard refresh: `Ctx::step` and `set_queue_step` each emitted a `rebuild_queue_changed` snapshot when the label changed, and those emissions go away with them. This is safe — the snapshot payload has no step field, so those pushes carried nothing a client could observe. Real state transitions still emit from the scheduler's claim and completion paths, from `submit`, and from the three `actions.rs` sites. Net effect is strictly fewer redundant SSE pushes. Docs: `docs/coordinator.md` still listed `step` as a `NodeView` wire field and `docs/web-ui/dashboard.md` documented a cyan `↳ <step>` sub-line under each queue row. Neither has existed since #2661 — both corrected here, plus the `job_queue/model.rs` module doc. Not touched: `frontend/packages/dashboard/src/system-sections.css` has a dead `.rqe-step` rule with no JS referencing it. Left for the frontend owner rather than deleted here. Closes: #2664
This commit is contained in:
atlas
mara
parent
ff62bf2235
commit
1db3cc32a1
9 changed files with 51 additions and 210 deletions
|
|
@ -20,7 +20,7 @@ fast-lane follow-up, the meta-update cascade pre-enqueue — are all just DAG
|
|||
|
||||
The **DAG** is the unit of cancel / approval-resolution and the
|
||||
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. Deps are intra-DAG edges only (`AfterOk` by default:
|
||||
the dep must succeed, a failed/cancelled dep cancels the dependent —
|
||||
cancel-downstream). Cross-DAG ordering comes from the per-agent lease,
|
||||
never from edges between DAGs. Submit-time validation (petgraph `toposort`)
|
||||
|
|
@ -167,7 +167,7 @@ without a row are seeded from observed state on first touch (running ⇒
|
|||
|
||||
The admin-socket responses carry the submitted DAG ids; `hivectl` polls
|
||||
`HostRequest::QueueDag` (~1s) and prints a progress line per DAG — roll-up
|
||||
glyph, template, agent, node chain with the running node's step label — so
|
||||
glyph, template, agent, node chain — so
|
||||
CLI verbs block until their jobs finish (`--no-wait` opts out; failures exit
|
||||
non-zero). Nodes appended in-DAG (a `MetaLock` growing per-agent rebuild
|
||||
subgraphs, a `Reconcile` fanning its `Start`/`Stop`) join the same DAG, so
|
||||
|
|
@ -248,11 +248,12 @@ cancelled-while-queued, which fails the approval instead of dangling it).
|
|||
`DagView` carries the entry-level fields (`id`, `kind` = template string,
|
||||
roll-up `state`, `source`, `reason`, timestamps, `inputs`,
|
||||
`approval_id`) plus `nodes: [NodeView…]` — per-node `agent`, `kind`, `deps`,
|
||||
`state`, `step`, `build_log_id`, timestamps, `error`. There is **no
|
||||
`state`, `build_log_id`, timestamps, `error`. There is **no
|
||||
DAG-level `agent`** (agent is per-node, so a DAG can span agents); consumers
|
||||
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.
|
||||
derive a DAG's agent(s) from its nodes. The node kind *is* the phase label —
|
||||
there is no separate sub-step string; 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.
|
||||
|
||||
---
|
||||
|
||||
|
|
|
|||
|
|
@ -220,12 +220,10 @@ op for the same agent collapses into the existing entry. All timing
|
|||
labels stay live: running entries tick elapsed seconds every second;
|
||||
queued and terminal ("done N ago" / "failed N ago") labels tick every
|
||||
30s so keyed rows never show stale timestamps as they persist across
|
||||
`rebuild_queue_changed` snapshots. When the worker has annotated the
|
||||
current phase a cyan `↳ <step>` sub-line appears under the main row
|
||||
showing the in-flight step name (e.g. `↳ meta prepare_deploy` →
|
||||
`↳ nixos-container update` → `↳ finalize deploy`). Terminal
|
||||
transitions clear `step` on the backend so Done / Failed rows don't
|
||||
render stale labels.
|
||||
`rebuild_queue_changed` snapshots. The in-flight phase is the running
|
||||
**node's kind** in the node chain — there is no separate sub-step
|
||||
label (the old cyan `↳ <step>` sub-line went away with the DAG queue:
|
||||
each phase is its own node now).
|
||||
Queued entries carry a `✗` cancel button on the right edge;
|
||||
running / done / failed / cancelled entries don't show it — the
|
||||
backend refuses cancellation for non-`Queued` rows anyway
|
||||
|
|
@ -242,7 +240,7 @@ It's keyed to the running entry's `build_log_id` and opens one
|
|||
`EventSource` to `GET /api/build-logs/id/{id}/stream` (the same stream
|
||||
the BUILD L0GS tab uses; the stream replays accumulated output on
|
||||
connect). It lives in its own container outside `#rebuild-queue-section`
|
||||
so the queue's per-row re-render (rows rebuild as the `step` advances)
|
||||
so the queue's per-row re-render (rows rebuild as nodes advance)
|
||||
never tears down the open stream; it hides when nothing is building and
|
||||
each row keeps its `logs →` link out to the full build log history.
|
||||
|
||||
|
|
@ -912,7 +910,7 @@ agent is stale. Banner pulses on each broker SSE event
|
|||
`queued` / `running` entries, a compact amber banner sits above the
|
||||
container list: `◐ build queue — N running · M queued — view queue →`
|
||||
(the link goes to the BU1LDS page's R3BU1LD QU3U3). It replaces the
|
||||
old per-transient spinner list; the actual running step for each
|
||||
old per-transient spinner list; the actual running node for each
|
||||
agent is already shown on its card (transient + in-flight-queue
|
||||
badges), so the top of the tab only needs the at-a-glance summary.
|
||||
|
||||
|
|
|
|||
|
|
@ -211,17 +211,12 @@ fn deploy_ctx(coord: &Coordinator, approval_id: i64) -> Result<DeployCtx> {
|
|||
/// fails, or if the eval-verify of the reviewed commit fails. Every one of
|
||||
/// these leaves the forge and `main` untouched, so the node is safely
|
||||
/// retryable.
|
||||
pub async fn run_deploy_merge_verify(
|
||||
coord: &Arc<Coordinator>,
|
||||
queue_entry_id: Option<u64>,
|
||||
approval_id: i64,
|
||||
) -> Result<()> {
|
||||
pub async fn run_deploy_merge_verify(coord: &Arc<Coordinator>, approval_id: i64) -> Result<()> {
|
||||
let ctx = deploy_ctx(coord, approval_id)?;
|
||||
let pr = ctx.pr;
|
||||
let reviewed = ctx.reviewed.as_str();
|
||||
|
||||
// 1. Drift gate: the live PR head must still equal what was reviewed.
|
||||
coord.set_queue_step(queue_entry_id, "verify PR head");
|
||||
let head = crate::forge::pr_head_sha(&ctx.repo, pr)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("read PR #{pr} head: {e}"))?;
|
||||
|
|
@ -232,13 +227,11 @@ pub async fn run_deploy_merge_verify(
|
|||
}
|
||||
|
||||
// 2. Fetch the reviewed head into applied so ff/verify/deploy resolve it.
|
||||
coord.set_queue_step(queue_entry_id, "fetch PR head");
|
||||
crate::forge::fetch_pr_head_into_applied(&ctx.repo, pr)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("fetch PR #{pr} head into applied: {e}"))?;
|
||||
|
||||
// 3. Eval-verify BEFORE the irreversible merge (bad nix fails fast here).
|
||||
coord.set_queue_step(queue_entry_id, "verify proposal (eval)");
|
||||
crate::meta::verify_commit(ctx.approval.agent.as_str(), &ctx.applied_dir, reviewed)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("verify merge head {reviewed}: {e:#}"))?;
|
||||
|
|
@ -264,11 +257,7 @@ pub async fn run_deploy_merge_verify(
|
|||
/// fast-forwarding `applied/main` / `meta::prepare_deploy` fails. From the merge
|
||||
/// onward a failure is *not* retryable on its own — [`run_deploy_tail`] runs
|
||||
/// `AfterAny` to compensate.
|
||||
pub async fn run_deploy_apply(
|
||||
coord: &Arc<Coordinator>,
|
||||
queue_entry_id: Option<u64>,
|
||||
approval_id: i64,
|
||||
) -> Result<()> {
|
||||
pub async fn run_deploy_apply(coord: &Arc<Coordinator>, approval_id: i64) -> Result<()> {
|
||||
let ctx = deploy_ctx(coord, approval_id)?;
|
||||
let agent = ctx.approval.agent.as_str();
|
||||
let pr = ctx.pr;
|
||||
|
|
@ -285,7 +274,6 @@ pub async fn run_deploy_apply(
|
|||
// both advances `main` to the reviewed head and marks the PR merged — no
|
||||
// direct push to the protected branch. A failure here means `main` was NOT
|
||||
// advanced, so it's fatal: we must not deploy a head the forge didn't merge.
|
||||
coord.set_queue_step(queue_entry_id, "fast-forward-merge PR");
|
||||
match crate::forge::merge_config_pr_ff(&ctx.repo, pr, &ctx.reviewed).await {
|
||||
Ok(()) => {}
|
||||
Err(crate::forge::ForgeMergeError::HeadDrift { expected, actual }) => bail!(
|
||||
|
|
@ -294,14 +282,7 @@ pub async fn run_deploy_apply(
|
|||
Err(e) => bail!("ff-merge PR #{pr}: {e}"),
|
||||
}
|
||||
|
||||
prepare_applied_target(
|
||||
coord,
|
||||
agent,
|
||||
&ctx.applied_dir,
|
||||
&ctx.reviewed,
|
||||
queue_entry_id,
|
||||
)
|
||||
.await
|
||||
prepare_applied_target(agent, &ctx.applied_dir, &ctx.reviewed).await
|
||||
}
|
||||
|
||||
/// `DeployTail` node body — compensation + bookkeeping, `AfterAny` the apply
|
||||
|
|
@ -371,7 +352,6 @@ pub async fn run_deploy_tail(
|
|||
}
|
||||
}
|
||||
|
||||
coord.set_queue_step(queue_entry_id, "roll back applied/main");
|
||||
if let Err(e) =
|
||||
lifecycle::git_update_ref(&applied_dir, "refs/heads/main", &prev_main).await
|
||||
{
|
||||
|
|
@ -389,7 +369,6 @@ pub async fn run_deploy_tail(
|
|||
}
|
||||
}
|
||||
|
||||
coord.set_queue_step(queue_entry_id, "forge push");
|
||||
if let Err(e) = crate::forge::push_config(agent).await {
|
||||
tracing::warn!(%agent, error = ?e, "forge: push_config after merge failed");
|
||||
}
|
||||
|
|
@ -789,13 +768,10 @@ fn finish_approval(
|
|||
/// (a merge is never a first spawn), so there's no `sync_agents` step — the
|
||||
/// operator `Spawn` flow owns first-time meta registration.
|
||||
async fn prepare_applied_target(
|
||||
coord: &Arc<Coordinator>,
|
||||
agent: &str,
|
||||
applied_dir: &std::path::Path,
|
||||
target: &str,
|
||||
queue_entry_id: Option<u64>,
|
||||
) -> Result<()> {
|
||||
coord.set_queue_step(queue_entry_id, "fast-forward applied/main");
|
||||
// Fast-forward applied/main to target + sync the working tree. Meta input
|
||||
// pins `?ref=main`, so this is what makes nix re-lock to the target commit
|
||||
// on the prepare_deploy step below.
|
||||
|
|
@ -806,7 +782,6 @@ async fn prepare_applied_target(
|
|||
.await
|
||||
.map_err(|e| anyhow::anyhow!("read-tree to main: {e:#}"))?;
|
||||
|
||||
coord.set_queue_step(queue_entry_id, "meta prepare_deploy");
|
||||
// Phase 1 of the meta two-phase deploy: relock without committing. The
|
||||
// staged lock then stays uncommitted across the whole appended rebuild —
|
||||
// which is why the `MetaWindow` is held by the deploy root, not by a node.
|
||||
|
|
@ -835,27 +810,20 @@ async fn prepare_applied_target(
|
|||
/// `meta::finalize_deploy` is deliberately *not* an error: the container is
|
||||
/// already running the new config by then, and the staged `flake.lock` it
|
||||
/// couldn't commit is something the operator can land by hand.
|
||||
pub async fn run_finalize_deploy(
|
||||
coord: &Arc<Coordinator>,
|
||||
queue_entry_id: Option<u64>,
|
||||
approval_id: i64,
|
||||
) -> Result<()> {
|
||||
pub async fn run_finalize_deploy(coord: &Arc<Coordinator>, approval_id: i64) -> Result<()> {
|
||||
let ctx = deploy_ctx(coord, approval_id)?;
|
||||
let agent = ctx.approval.agent.as_str();
|
||||
let target = ctx.reviewed.as_str();
|
||||
|
||||
coord.set_queue_step(queue_entry_id, "drop rollback ref");
|
||||
lifecycle::git_delete_ref(&ctx.applied_dir, &rollback_ref(approval_id))
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("drop rollback ref for approval {approval_id}: {e:#}"))?;
|
||||
|
||||
coord.set_queue_step(queue_entry_id, "plant deployed tag");
|
||||
let tag = format!("deployed/{approval_id}");
|
||||
lifecycle::git_tag(&ctx.applied_dir, &tag, target)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("plant {tag}: {e:#}"))?;
|
||||
|
||||
coord.set_queue_step(queue_entry_id, "meta finalize_deploy");
|
||||
if let Err(e) = crate::meta::finalize_deploy(agent, target, &tag).await {
|
||||
tracing::warn!(%agent, approval_id, error = ?e, "meta finalize_deploy failed");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -666,23 +666,6 @@ impl Coordinator {
|
|||
});
|
||||
}
|
||||
|
||||
/// Update the `step` label on the currently-running node of DAG
|
||||
/// `id` and (if it actually changed) re-emit the queue snapshot so
|
||||
/// the dashboard renders the new phase. DAG-id-only surface for the
|
||||
/// approval-deploy bodies in `actions.rs`, which don't know their node id.
|
||||
/// The lookup is still exact for them: a deploy DAG is a strictly
|
||||
/// sequential chain whose resource-holding root sits in `Finishing` while
|
||||
/// the phases run, so at most one node is ever `Running`. Queue executors
|
||||
/// that do know their node id use the precise per-node sink in
|
||||
/// `job_queue::exec` instead. No-op when `id` is `None` (callers not
|
||||
/// running from the queue) or when nothing is `Running`.
|
||||
pub fn set_queue_step(self: &Arc<Self>, id: Option<u64>, step: &str) {
|
||||
let Some(id) = id else { return };
|
||||
if self.job_queue.set_step_running(id, step) {
|
||||
self.emit_rebuild_queue_snapshot();
|
||||
}
|
||||
}
|
||||
|
||||
/// Subscribe to the shutdown watch channel. Background tasks call
|
||||
/// this at spawn time and break their loop when the receiver
|
||||
/// transitions to `true` (via `Coordinator::request_shutdown`).
|
||||
|
|
|
|||
|
|
@ -43,7 +43,7 @@ pub struct NodeOutput {
|
|||
pub append_subgraph: Vec<Vec<NodeSpec>>,
|
||||
}
|
||||
|
||||
/// Step-label + build-log sink for one claimed node.
|
||||
/// Build-log sink for one claimed node.
|
||||
struct Ctx<'a> {
|
||||
coord: &'a Arc<Coordinator>,
|
||||
dag_id: u64,
|
||||
|
|
@ -51,16 +51,6 @@ struct Ctx<'a> {
|
|||
}
|
||||
|
||||
impl Ctx<'_> {
|
||||
fn step(&self, step: &str) {
|
||||
if self
|
||||
.coord
|
||||
.job_queue
|
||||
.set_step(self.dag_id, self.node_id, step)
|
||||
{
|
||||
self.coord.emit_rebuild_queue_snapshot();
|
||||
}
|
||||
}
|
||||
|
||||
fn build_log(&self, log_id: i64) {
|
||||
if self
|
||||
.coord
|
||||
|
|
@ -85,20 +75,20 @@ pub(super) async fn run_node(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
|
|||
NodeKind::MetaSync { relock, .. } => run_meta_sync(coord, claim, *relock).await,
|
||||
NodeKind::Prebuild { .. } => run_prebuild(claim, &ctx).await,
|
||||
NodeKind::Swap { .. } => run_swap(coord, claim, &ctx).await,
|
||||
NodeKind::PostSwap { .. } => run_post_swap(coord, claim, &ctx).await,
|
||||
NodeKind::Provision { .. } => run_provision(coord, claim, &ctx).await,
|
||||
NodeKind::Create { .. } => run_create(claim, &ctx).await,
|
||||
NodeKind::PostSwap { .. } => run_post_swap(coord, claim).await,
|
||||
NodeKind::Provision { .. } => run_provision(coord, claim).await,
|
||||
NodeKind::Create { .. } => run_create(claim).await,
|
||||
NodeKind::MetaLock { sweep, fanout } => {
|
||||
run_meta_lock(coord, claim, &ctx, *sweep, fanout.clone()).await
|
||||
run_meta_lock(coord, claim, *sweep, fanout.clone()).await
|
||||
}
|
||||
NodeKind::Reconcile { .. } => run_reconcile(coord, claim).await,
|
||||
NodeKind::Start { .. } => run_start(coord, claim, &ctx).await,
|
||||
NodeKind::Stop { .. } => run_stop(coord, claim, &ctx).await,
|
||||
NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, claim, &ctx).await,
|
||||
NodeKind::Signal { .. } => Ok(run_signal(coord, claim, &ctx)),
|
||||
NodeKind::Drain { .. } => run_drain(coord, claim, &ctx).await,
|
||||
NodeKind::Start { .. } => run_start(coord, claim).await,
|
||||
NodeKind::Stop { .. } => run_stop(coord, claim).await,
|
||||
NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, claim).await,
|
||||
NodeKind::Signal { .. } => Ok(run_signal(coord, claim)),
|
||||
NodeKind::Drain { .. } => run_drain(coord, claim).await,
|
||||
NodeKind::WriteDropin { .. } => run_write_dropin(coord, claim).await,
|
||||
NodeKind::WritePermFile { .. } => run_write_perm_file(coord, claim, &ctx).await,
|
||||
NodeKind::WritePermFile { .. } => run_write_perm_file(coord, claim).await,
|
||||
NodeKind::DeployWindow { .. } => run_deploy_window(claim),
|
||||
NodeKind::MergeVerify { .. } => run_merge_verify(coord, claim).await,
|
||||
NodeKind::DeployApply { .. } => run_deploy_apply(coord, claim).await,
|
||||
|
|
@ -239,12 +229,9 @@ async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
|
|||
// stopped agent has no uptime to preserve, so skip the (expensive)
|
||||
// eval and let the downstream `Swap` build inline.
|
||||
if crate::lifecycle::is_running(name).await {
|
||||
ctx.step("nix build");
|
||||
let flake_ref = format!("{}#{name}", crate::paths::meta_root().display());
|
||||
crate::lifecycle::prebuild_toplevel(name, &flake_ref, &|log_id| ctx.build_log(log_id))
|
||||
.await?;
|
||||
} else {
|
||||
ctx.step("skipped prebuild (container down)");
|
||||
}
|
||||
Ok(NodeOutput::default())
|
||||
}
|
||||
|
|
@ -261,11 +248,10 @@ async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Res
|
|||
let agent_dir = crate::paths::agent_runtime_dir(name);
|
||||
let hive = coord.hive_env();
|
||||
let paths = Coordinator::agent_paths(name, agent_dir);
|
||||
let result =
|
||||
crate::lifecycle::swap_update(name, &hive, &paths, &|step| ctx.step(step), &|log_id| {
|
||||
ctx.build_log(log_id);
|
||||
})
|
||||
.await;
|
||||
let result = crate::lifecycle::swap_update(name, &hive, &paths, &|log_id| {
|
||||
ctx.build_log(log_id);
|
||||
})
|
||||
.await;
|
||||
// On success the Ok-only bookkeeping tail (rev marker, forge/matrix
|
||||
// sync, kick, rescan, snapshot) runs in the sibling `PostSwap` node,
|
||||
// which deps `AfterOk(Swap)`. On failure `PostSwap` is cancel-cascaded
|
||||
|
|
@ -284,11 +270,7 @@ async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Res
|
|||
/// means the profile swap succeeded. Store/forge/matrix work only — no nix
|
||||
/// build (build-slot-exempt); the agent lease taken at `Swap` is still held
|
||||
/// (the whole chain up to `Reconcile` is one agent's subgraph).
|
||||
async fn run_post_swap(
|
||||
coord: &Arc<Coordinator>,
|
||||
claim: &Claim,
|
||||
ctx: &Ctx<'_>,
|
||||
) -> Result<NodeOutput> {
|
||||
async fn run_post_swap(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
||||
let name = &claim.agent;
|
||||
if let Some(rev) = crate::auto_update::current_flake_rev(&coord.hyperhive_flake)
|
||||
&& let Err(e) = std::fs::write(crate::paths::applied_rev_marker(name), rev)
|
||||
|
|
@ -298,7 +280,6 @@ async fn run_post_swap(
|
|||
// The `Rebuilt` manager event fires exactly once per DAG from the
|
||||
// terminal hook — emitting ok here and letting a failed tail `Reconcile`
|
||||
// add a contradictory !ok would double-report the same rebuild.
|
||||
ctx.step("forge sync");
|
||||
// Full forge + matrix sync on every successful rebuild so the rebuild
|
||||
// path is equivalent to the startup sweep: tokens, config-repo mirror,
|
||||
// meta access all recover without a hive-c0re restart.
|
||||
|
|
@ -317,16 +298,11 @@ async fn run_post_swap(
|
|||
/// subvolume, and the meta `sync_agents` registration. Runs under the
|
||||
/// deploy window (`NodeKind::needs_meta_window`) so its commit can't
|
||||
/// land inside another node's staged deploy window.
|
||||
async fn run_provision(
|
||||
coord: &Arc<Coordinator>,
|
||||
claim: &Claim,
|
||||
ctx: &Ctx<'_>,
|
||||
) -> Result<NodeOutput> {
|
||||
async fn run_provision(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
||||
let name = &claim.agent;
|
||||
let agent_dir = crate::paths::agent_runtime_dir(name);
|
||||
let hive = coord.hive_env();
|
||||
let paths = Coordinator::agent_paths(name, agent_dir);
|
||||
ctx.step("provisioning");
|
||||
crate::lifecycle::provision_container(name, &hive, &paths).await?;
|
||||
Ok(NodeOutput::default())
|
||||
}
|
||||
|
|
@ -337,8 +313,7 @@ async fn run_provision(
|
|||
/// dir creation and MCP listener registration are deferred to the tail
|
||||
/// `Reconcile` (`converge_start_preamble` + `register_agent`) so this
|
||||
/// node stays purely "create", not "create + start".
|
||||
async fn run_create(claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
|
||||
ctx.step("nixos-container create");
|
||||
async fn run_create(claim: &Claim) -> Result<NodeOutput> {
|
||||
crate::lifecycle::create_only(&claim.agent).await?;
|
||||
Ok(NodeOutput::default())
|
||||
}
|
||||
|
|
@ -350,12 +325,10 @@ async fn run_create(claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
|
|||
async fn run_meta_lock(
|
||||
coord: &Arc<Coordinator>,
|
||||
claim: &Claim,
|
||||
ctx: &Ctx<'_>,
|
||||
sweep: bool,
|
||||
fanout: Option<Vec<String>>,
|
||||
) -> Result<NodeOutput> {
|
||||
if sweep {
|
||||
ctx.step("nix flake update hyperhive");
|
||||
if let Err(e) = crate::meta::lock_update_hyperhive().await {
|
||||
tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed");
|
||||
}
|
||||
|
|
@ -371,7 +344,6 @@ async fn run_meta_lock(
|
|||
return Ok(NodeOutput { append_subgraph });
|
||||
}
|
||||
let _progress = coord.meta_update_guard();
|
||||
ctx.step("nix flake update");
|
||||
crate::meta::lock_update(&claim.inputs).await?;
|
||||
// Lock file changed — meta-inputs panel re-renders.
|
||||
crate::dashboard::emit_meta_inputs_snapshot(coord);
|
||||
|
|
@ -424,7 +396,7 @@ async fn run_reconcile(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOu
|
|||
|
||||
/// Mechanical container start — the sub-step a `Reconcile` planner fans
|
||||
/// out when it observes `wanted = Up` and the container down.
|
||||
async fn run_start(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
|
||||
async fn run_start(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
||||
let name = &claim.agent;
|
||||
// Node-local transient only when the DAG holds none (the
|
||||
// boot-reconcile template); a rebuild/spawn/etc. DAG's lease-window
|
||||
|
|
@ -441,7 +413,6 @@ async fn run_start(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Re
|
|||
let hive = coord.hive_env();
|
||||
let paths = Coordinator::agent_paths(name, agent_dir);
|
||||
let token = crate::lifecycle::converge_start_preamble(name, &hive, &paths).await?;
|
||||
ctx.step("nixos-container start");
|
||||
crate::lifecycle::start_with_fallback(token).await?;
|
||||
// Bind the MCP listener immediately after starting the container.
|
||||
// The preamble created the runtime dir; the container is now coming
|
||||
|
|
@ -456,13 +427,12 @@ async fn run_start(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Re
|
|||
|
||||
/// Mechanical container stop — the sub-step a `Reconcile` planner fans
|
||||
/// out when it observes `wanted = Offline` and the container up.
|
||||
async fn run_stop(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
|
||||
async fn run_stop(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
||||
let name = &claim.agent;
|
||||
let _guard = claim
|
||||
.transient
|
||||
.is_none()
|
||||
.then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Stopping));
|
||||
ctx.step("nixos-container stop");
|
||||
crate::lifecycle::kill(name).await?;
|
||||
coord.unregister_agent(name);
|
||||
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
|
||||
|
|
@ -474,11 +444,7 @@ async fn run_stop(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Res
|
|||
|
||||
/// Mechanical stop for the profile swap. Never *changes* `wanted`;
|
||||
/// noop when already stopped.
|
||||
async fn run_stop_for_update(
|
||||
coord: &Arc<Coordinator>,
|
||||
claim: &Claim,
|
||||
ctx: &Ctx<'_>,
|
||||
) -> Result<NodeOutput> {
|
||||
async fn run_stop_for_update(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
||||
let name = &claim.agent;
|
||||
if crate::lifecycle::is_running(name).await {
|
||||
// Seed a missing agent_power row from the PRE-stop observation
|
||||
|
|
@ -488,7 +454,6 @@ async fn run_stop_for_update(
|
|||
if let Err(e) = coord.power.get_or_seed(name, true) {
|
||||
tracing::warn!(%name, error = ?e, "agent_power: pre-stop seed failed");
|
||||
}
|
||||
ctx.step("nixos-container stop");
|
||||
crate::lifecycle::kill(name).await?;
|
||||
coord.rescan_containers_and_emit().await;
|
||||
}
|
||||
|
|
@ -504,12 +469,10 @@ async fn run_stop_for_update(
|
|||
/// `GRACEFUL_STOP_TIMEOUT`. Safe because the harness tests the marker at
|
||||
/// the top of its loop — a paused agent has no turn in flight, so there
|
||||
/// is nothing to checkpoint.
|
||||
fn run_signal(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> NodeOutput {
|
||||
fn run_signal(coord: &Arc<Coordinator>, claim: &Claim) -> NodeOutput {
|
||||
if hive_types::Ident::parse(&claim.agent).is_ok_and(|a| Coordinator::is_paused(&a)) {
|
||||
ctx.step("graceful stop: agent paused, nothing to drain");
|
||||
return NodeOutput::default();
|
||||
}
|
||||
ctx.step("graceful stop: signalling agent");
|
||||
coord.mark_graceful_stop(&claim.agent);
|
||||
coord.kick_agent(&claim.agent, "graceful stop requested");
|
||||
NodeOutput::default()
|
||||
|
|
@ -518,9 +481,8 @@ fn run_signal(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> NodeOut
|
|||
/// Await the harness clearing the fence (`GracefulStopComplete`) or
|
||||
/// the timeout — either way the downstream `Reconcile` proceeds with
|
||||
/// the actual stop.
|
||||
async fn run_drain(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
|
||||
async fn run_drain(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
||||
let name = &claim.agent;
|
||||
ctx.step("graceful stop: draining");
|
||||
let deadline = std::time::Instant::now() + GRACEFUL_STOP_TIMEOUT;
|
||||
while coord.is_graceful_stop_pending(name) {
|
||||
if std::time::Instant::now() >= deadline {
|
||||
|
|
@ -549,18 +511,13 @@ async fn run_write_dropin(coord: &Arc<Coordinator>, claim: &Claim) -> Result<Nod
|
|||
/// Write + commit the perm file(s) (fused under `META_LOCK` so the
|
||||
/// working tree is never left dirty), then emit the P3RM1SS10NS-tab
|
||||
/// snapshots so the dashboard reflects the new assignment.
|
||||
async fn run_write_perm_file(
|
||||
coord: &Arc<Coordinator>,
|
||||
claim: &Claim,
|
||||
ctx: &Ctx<'_>,
|
||||
) -> Result<NodeOutput> {
|
||||
async fn run_write_perm_file(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
||||
use super::model::PermPayload;
|
||||
let name = &claim.agent;
|
||||
// The perm file payload rides the node itself (the only consumer).
|
||||
let NodeKind::WritePermFile { payload, .. } = &claim.kind else {
|
||||
anyhow::bail!("run_write_perm_file on a non-WritePermFile node");
|
||||
};
|
||||
ctx.step("writing + committing perm file");
|
||||
// Runs under the deploy window (`NodeKind::needs_meta_window`): a
|
||||
// perm commit landing inside another node's staged prepare→finalize
|
||||
// window would sweep the staged deploy lock into its commit (the
|
||||
|
|
@ -622,7 +579,7 @@ fn run_deploy_window(claim: &Claim) -> Result<NodeOutput> {
|
|||
/// failure here cancel-cascades the rest of the subtree with the forge and the
|
||||
/// applied repo exactly as they were.
|
||||
async fn run_merge_verify(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
||||
crate::actions::run_deploy_merge_verify(coord, Some(claim.dag_id), deploy_approval_id(claim)?)
|
||||
crate::actions::run_deploy_merge_verify(coord, deploy_approval_id(claim)?)
|
||||
.await
|
||||
.map(|()| NodeOutput::default())
|
||||
}
|
||||
|
|
@ -637,7 +594,7 @@ async fn run_merge_verify(coord: &Arc<Coordinator>, claim: &Claim) -> Result<Nod
|
|||
/// ancestor already holding it. On failure nothing is appended and the tail
|
||||
/// compensates, exactly as before.
|
||||
async fn run_deploy_apply(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
||||
crate::actions::run_deploy_apply(coord, Some(claim.dag_id), deploy_approval_id(claim)?).await?;
|
||||
crate::actions::run_deploy_apply(coord, deploy_approval_id(claim)?).await?;
|
||||
Ok(NodeOutput {
|
||||
append_subgraph: vec![super::templates::deploy_rebuild_nodes(claim.kind.agent())],
|
||||
})
|
||||
|
|
@ -647,7 +604,7 @@ async fn run_deploy_apply(coord: &Arc<Coordinator>, claim: &Claim) -> Result<Nod
|
|||
/// come up clean: drop the rollback ref, plant the `deployed/<id>` tag, commit
|
||||
/// the staged lock.
|
||||
async fn run_finalize_deploy(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
||||
crate::actions::run_finalize_deploy(coord, Some(claim.dag_id), deploy_approval_id(claim)?)
|
||||
crate::actions::run_finalize_deploy(coord, deploy_approval_id(claim)?)
|
||||
.await
|
||||
.map(|()| NodeOutput::default())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -99,11 +99,10 @@ pub struct TerminalDag {
|
|||
|
||||
/// Per-node runtime metadata the crate graph doesn't carry. Lifecycle
|
||||
/// (`started_at` / `finished_at` / `error`) lives on the `hive_jobq::Node`
|
||||
/// itself now, so only the two host-side extras remain: the live sub-step
|
||||
/// label and the build-log row link (the client fetches the log by node id).
|
||||
/// itself now, so only the build-log row link remains host-side (the
|
||||
/// client fetches the log by node id).
|
||||
#[derive(Debug, Default, Clone)]
|
||||
struct NodeRuntime {
|
||||
step: Option<String>,
|
||||
build_log_id: Option<i64>,
|
||||
}
|
||||
|
||||
|
|
@ -129,8 +128,8 @@ struct DagMeta {
|
|||
/// [`QueueInner::dag_meta`]). One shared crate [`Graph`] holds every DAG.
|
||||
struct QueueInner {
|
||||
sched: Scheduler<NodeKind, Resource>,
|
||||
/// Per-node runtime metadata (build-log id, step, timestamps, error) —
|
||||
/// mutable after insert, so it can't ride the immutable node payload.
|
||||
/// Per-node runtime metadata (the build-log id) — mutable after
|
||||
/// insert, so it can't ride the immutable node payload.
|
||||
node_rt: HashMap<NodeId, NodeRuntime>,
|
||||
}
|
||||
|
||||
|
|
@ -405,14 +404,11 @@ impl JobQueue {
|
|||
let mut inner = self.lock();
|
||||
// The failure reason + `finished_at` are stamped onto the graph `Node`
|
||||
// by the scheduler (the reason rides `Outcome::Failed`); no host-side
|
||||
// copy. We only clear the live sub-step label here.
|
||||
// copy, so there is nothing to clear here.
|
||||
let outcome = match result {
|
||||
Ok(()) => Outcome::Done,
|
||||
Err(e) => Outcome::Failed(truncate_error(&e)),
|
||||
};
|
||||
if let Some(rt) = inner.node_rt.get_mut(&node_id) {
|
||||
rt.step = None;
|
||||
}
|
||||
let container = inner.dag_of(node_id);
|
||||
inner.sched.complete(node_id, outcome);
|
||||
// If this completion rolled the DAG's container up to a terminal state,
|
||||
|
|
@ -461,35 +457,6 @@ impl JobQueue {
|
|||
terminal
|
||||
}
|
||||
|
||||
/// Set the step label on a `Running` node. Returns `true` when it changed.
|
||||
pub fn set_step(&self, dag_id: u64, node_id: NodeId, step: &str) -> bool {
|
||||
let mut inner = self.lock();
|
||||
if inner.dag_of(node_id).map(NodeId::get) != Some(dag_id) || !inner.node_running(node_id) {
|
||||
return false;
|
||||
}
|
||||
let rt = inner.node_rt.entry(node_id).or_default();
|
||||
if rt.step.as_deref() == Some(step) {
|
||||
return false;
|
||||
}
|
||||
rt.step = Some(step.to_owned());
|
||||
true
|
||||
}
|
||||
|
||||
/// Set the step label on the DAG's currently-running node — the DAG-id-only
|
||||
/// compatibility surface for the opaque approval pipeline.
|
||||
pub fn set_step_running(&self, dag_id: u64, step: &str) -> bool {
|
||||
let mut inner = self.lock();
|
||||
let Some(node_id) = inner.running_node_of(dag_id) else {
|
||||
return false;
|
||||
};
|
||||
let rt = inner.node_rt.entry(node_id).or_default();
|
||||
if rt.step.as_deref() == Some(step) {
|
||||
return false;
|
||||
}
|
||||
rt.step = Some(step.to_owned());
|
||||
true
|
||||
}
|
||||
|
||||
/// Link a `build_logs` row to a specific `Running` node.
|
||||
pub fn set_build_log_id(&self, dag_id: u64, node_id: NodeId, log_id: i64) -> bool {
|
||||
let mut inner = self.lock();
|
||||
|
|
@ -670,15 +637,6 @@ impl QueueInner {
|
|||
})
|
||||
}
|
||||
|
||||
/// The DAG's currently-running work node, if any (the opaque approval
|
||||
/// pipeline's single-node DAGs make this exact).
|
||||
fn running_node_of(&self, dag_id: u64) -> Option<NodeId> {
|
||||
let container = self.container(dag_id)?;
|
||||
self.subtree(container)
|
||||
.into_iter()
|
||||
.find(|&id| self.node_running(id))
|
||||
}
|
||||
|
||||
/// Roll-up state over a DAG's work nodes: `Failed` if any failed; else
|
||||
/// `Running` if any running; else `Queued` if any queued; else `Cancelled`
|
||||
/// if any cancelled; else `Done`. (Kept eager over the subtree — a failed
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@
|
|||
//!
|
||||
//! Two levels: the **DAG** is the unit of cancel / approval-resolution
|
||||
//! and the dashboard group; the **node** is the unit of scheduling /
|
||||
//! execution / build-log / step label, and carries its own `agent` (a
|
||||
//! execution / build-log, and carries its own `agent` (a
|
||||
//! DAG can span agents). See `docs/coordinator.md::Job queue` for the
|
||||
//! full design.
|
||||
|
||||
|
|
|
|||
|
|
@ -1134,29 +1134,7 @@ fn deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails() {
|
|||
assert_eq!(summary.approval_id, Some(9));
|
||||
}
|
||||
|
||||
// ---- steps, build logs, history ----
|
||||
|
||||
#[test]
|
||||
fn set_step_only_on_running_and_signals_change() {
|
||||
let q = JobQueue::new(1);
|
||||
let id = submit(&q, rebuild("agent-a", "r"));
|
||||
assert!(
|
||||
!q.set_step_running(id, "too early"),
|
||||
"no running node yet → refused"
|
||||
);
|
||||
let c = claim_one(&q);
|
||||
assert!(q.set_step(id, c.node_id, "nix build"));
|
||||
assert!(
|
||||
!q.set_step(id, c.node_id, "nix build"),
|
||||
"same label → false"
|
||||
);
|
||||
assert!(q.set_step(id, c.node_id, "next phase"));
|
||||
assert!(q.set_step_running(id, "via running lookup"));
|
||||
// `step` is host-side only now (off the wire); completion clears it
|
||||
// internally, but there's no wire field to observe — the return-value
|
||||
// contract above (running-gating + change signalling) is the behaviour.
|
||||
q.complete_node(id, c.node_id, Ok(()));
|
||||
}
|
||||
// ---- build logs, history ----
|
||||
|
||||
#[test]
|
||||
fn set_build_log_id_links_running_node() {
|
||||
|
|
|
|||
|
|
@ -273,11 +273,9 @@ pub async fn swap_update(
|
|||
name: &str,
|
||||
hive: &HiveEnv,
|
||||
paths: &AgentPaths,
|
||||
on_step: &(dyn Fn(&str) + Send + Sync),
|
||||
on_build_log_id: &(dyn Fn(i64) + Send + Sync),
|
||||
) -> Result<()> {
|
||||
write_dropins(name, hive, paths).await?;
|
||||
on_step("nixos-container update");
|
||||
priv_run_inner("update", name, Some(on_build_log_id)).await
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue