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

split Swap's Ok-tail into a first-class PostSwap DAG node (#2390)

Browse source
This commit is contained in:
damocles 2026-07-16 12:09:15 +02:00 • committed by mara
commit f2ff0deb6b
4 changed files with 126 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

75
hive-c0re/src/job_queue/exec.rs
View file

@ -84,6 +84,7 @@ pub(super) async fn run_node(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
match &claim.kind {
NodeKind::Prebuild { relock } => run_prebuild(coord, claim, &ctx, *relock).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::MetaLock { sweep, fanout } => {
@ -192,41 +193,53 @@ async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Res
ctx.build_log(log_id);
})
.await;
match &result {
Ok(()) => {
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)
{
tracing::warn!(%name, error = ?e, "write rev marker failed");
}
// 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.
crate::forge::sync_agent(name, crate::forge::core_token().as_deref()).await;
crate::matrix::sync_agent_standalone(name).await;
// Wake the agent on its next turn so claude sees a "you
// were rebuilt" hint; rescan so dashboards drop the
// "needs update" chip; lock bump → meta-inputs re-render.
coord.kick_agent(name, "container rebuilt");
coord.rescan_containers_and_emit().await;
crate::dashboard::emit_meta_inputs_snapshot(coord);
}
Err(_) => {
// The `Rebuilt { ok: false }` manager event fires once per
// DAG from the terminal hook (any node may be the one that
// failed); here only refresh the observed state.
coord.rescan_containers_and_emit().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
// and the tail `Reconcile` (`AfterAny(PostSwap)`) handles recovery; here
// we only refresh the observed state so dashboards reflect the failed
// swap immediately. The `Rebuilt { ok: false }` manager event fires once
// per DAG from the terminal hook (any node may be the one that failed).
if result.is_err() {
coord.rescan_containers_and_emit().await;
}
result.map(|()| NodeOutput::default())
}
/// The post-`Swap` bookkeeping tail, split into its own node for dashboard
/// visibility + retry granularity. Deps `AfterOk(Swap)`, so reaching here
/// 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> {
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)
{
tracing::warn!(%name, error = ?e, "write rev marker failed");
}
// 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.
crate::forge::sync_agent(name, crate::forge::core_token().as_deref()).await;
crate::matrix::sync_agent_standalone(name).await;
// Wake the agent on its next turn so claude sees a "you were rebuilt"
// hint; rescan so dashboards drop the "needs update" chip; lock bump →
// meta-inputs re-render.
coord.kick_agent(name, "container rebuilt");
coord.rescan_containers_and_emit().await;
crate::dashboard::emit_meta_inputs_snapshot(coord);
Ok(NodeOutput::default())
}
/// First-spawn pre-create provisioning: proposed/applied repos, state
/// subvolume, and the meta `sync_agents` registration. Holds the
/// deploy-window gate for the commit so it can't land inside another

15
hive-c0re/src/job_queue/model.rs
View file

@ -57,9 +57,19 @@ pub enum NodeKind {
Prebuild { relock: bool },
/// `nixos-container update` profile-swap (requires the container
/// stopped). Re-applies nspawn flags + resource limits first —
/// rebuild is the reconcile verb — and carries the post-rebuild
/// bookkeeping tail (rev marker, forge/matrix sync, kick, rescan).
/// rebuild is the reconcile verb. The post-rebuild bookkeeping tail
/// lives in the sibling `PostSwap` node.
Swap,
/// The post-`Swap` bookkeeping tail as a first-class node: rev marker,
/// forge + matrix sync, manager kick, container rescan, meta-inputs
/// snapshot. Split out of `Swap` for dashboard visibility + retry
/// granularity. Deps `AfterOk(Swap)`, so it runs only when the profile
/// swap succeeded; the tail `Reconcile` deps `AfterAny(PostSwap)`, so on
/// swap failure this node is cancel-cascaded (a terminal state) and
/// recovery still runs. Store/forge/matrix work only — no nix build, so
/// build-slot-exempt; the agent lease taken at `Swap` is held across the
/// whole chain until `Reconcile` settles, so it's not re-declared here.
PostSwap,
/// First-spawn pre-create provisioning: proposed/applied repos,
/// state subvolume, and meta registration (`sync_agents`). Runs
/// ahead of `Create` so the `nixos-container create --flake
@ -137,6 +147,7 @@ impl NodeKind {
match self {
NodeKind::Prebuild { .. } => "prebuild",
NodeKind::Swap => "swap",
NodeKind::PostSwap => "post_swap",
NodeKind::Provision => "provision",
NodeKind::Create => "create",
NodeKind::MetaLock { .. } => "meta_lock",

19
hive-c0re/src/job_queue/templates.rs
View file

@ -16,7 +16,7 @@
//! intent+reconcile is atomic per-agent).
//!
//! ```text
//! rebuild(a): Prebuild(a) → StopForUpdate(a) → Swap(a) →(any) Reconcile(a)
//! rebuild(a): Prebuild(a) → StopForUpdate(a) → Swap(a) →(ok) PostSwap(a) →(any) Reconcile(a)
//! spawn(a): Provision(a) → Create(a) → WriteDropin(a) → Reconcile(a) [wanted=Up at approve]
//! perm-change(a): WritePermFile(a) → «rebuild subgraph»
//! meta-update(inp): MetaLock(inp) →«in-DAG rebuild subgraph per affected a»
@ -50,10 +50,16 @@ pub(crate) fn node(agent: &str, kind: NodeKind, deps: Vec<Dep>) -> NodeSpec {
}
}
/// The rebuild node chain. `Reconcile` deps on `Swap` with `AfterAny`:
/// it must run even when the profile swap failed, so a previously-up
/// agent comes back on its old config (today's recovery-start). This
/// is the only `AfterAny` edge in v1.
/// The rebuild node chain. `PostSwap` carries the swap's Ok-only
/// bookkeeping tail (rev marker, forge/matrix sync, kick, rescan) and deps
/// `Swap` with `AfterOk`. `Reconcile` then deps on `PostSwap` with
/// `AfterAny`: it must run even when the swap failed, so a previously-up
/// agent comes back on its old config (today's recovery-start). On swap
/// failure the `AfterOk` `PostSwap` is cancel-cascaded to a terminal state,
/// which still satisfies `Reconcile`'s `AfterAny` edge — the only `AfterAny`
/// edge in v1. Pointing `Reconcile` at `PostSwap` (not `Swap`) also
/// serializes the tail ahead of the reconcile, so there's no double
/// rescan/kick race.
pub(crate) fn rebuild_nodes(agent: &str, relock: bool, base: u32) -> Vec<NodeSpec> {
vec![
node(
@ -67,11 +73,12 @@ pub(crate) fn rebuild_nodes(agent: &str, relock: bool, base: u32) -> Vec<NodeSpe
),
node(agent, NodeKind::StopForUpdate, after_ok(base)),
node(agent, NodeKind::Swap, after_ok(base + 1)),
node(agent, NodeKind::PostSwap, after_ok(base + 2)),
node(
agent,
NodeKind::Reconcile,
vec![Dep {
on: base + 2,
on: base + 3,
when: DepWhen::AfterAny,
}],
),

60
hive-c0re/src/job_queue/tests.rs
View file

@ -150,7 +150,13 @@ fn unknown_dep_is_rejected_at_submit() {
fn rebuild_chain_claims_in_dep_order() {
let q = JobQueue::new(1);
let id = submit(&q, rebuild("agent-a", "r"));
for expected in ["prebuild", "stop_for_update", "swap", "reconcile"] {
for expected in [
"prebuild",
"stop_for_update",
"swap",
"post_swap",
"reconcile",
] {
let c = claim_one(&q);
assert_eq!(c.dag_id, id);
assert_eq!(c.kind.as_str(), expected);
@ -624,6 +630,7 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() {
assert_eq!(by_kind("prebuild"), State::Failed);
assert_eq!(by_kind("stop_for_update"), State::Cancelled);
assert_eq!(by_kind("swap"), State::Cancelled);
assert_eq!(by_kind("post_swap"), State::Cancelled);
assert_eq!(by_kind("reconcile"), State::Done);
assert_eq!(
dag.nodes
@ -634,9 +641,10 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() {
);
}
/// The swap-failure recovery: `Swap` fails → the `AfterAny` edge still
/// runs `Reconcile`, which brings a wanted-up agent back on its old
/// config.
/// The swap-failure recovery: `Swap` fails → the `AfterOk` `PostSwap` is
/// cancel-cascaded → its terminal state still satisfies `Reconcile`'s
/// `AfterAny(PostSwap)` edge, so recovery-start runs and brings a wanted-up
/// agent back on its old config.
#[test]
fn swap_failure_still_runs_reconcile() {
let q = JobQueue::new(1);
@ -648,12 +656,55 @@ fn swap_failure_still_runs_reconcile() {
let swap = claim_one(&q);
assert_eq!(swap.kind.as_str(), "swap");
q.complete_node(id, swap.node_id, Err("update failed".to_owned()));
// PostSwap (AfterOk on the failed Swap) is cancel-cascaded; Reconcile is
// next-claimable via its AfterAny(PostSwap) edge.
let reconcile = claim_one(&q);
assert_eq!(reconcile.kind.as_str(), "reconcile");
q.complete_node(id, reconcile.node_id, Ok(()));
let all_dags = q.snapshot();
let dag = all_dags.iter().find(|d| d.id == id).expect("dag");
assert_eq!(
dag.nodes
.iter()
.find(|n| n.kind == "post_swap")
.expect("post_swap node")
.state,
State::Cancelled,
"PostSwap must cancel-cascade when Swap fails"
);
assert_eq!(state_of(&q, id), State::Failed);
}
/// The swap-success path: `Swap` ok → the `AfterOk` `PostSwap` (bookkeeping
/// tail) runs, and only then does `Reconcile` fire — serialized behind
/// `PostSwap` (not racing it) because `Reconcile` deps `AfterAny(PostSwap)`.
#[test]
fn swap_ok_runs_post_swap_before_reconcile() {
let q = JobQueue::new(1);
let id = submit(&q, rebuild("agent-a", "r"));
// prebuild + stop_for_update
for _ in 0..2 {
let c = claim_one(&q);
q.complete_node(id, c.node_id, Ok(()));
}
let swap = claim_one(&q);
assert_eq!(swap.kind.as_str(), "swap");
q.complete_node(id, swap.node_id, Ok(()));
// PostSwap runs next, and nothing else is claimable while it does — the
// tail serializes ahead of Reconcile.
let post_swap = claim_one(&q);
assert_eq!(post_swap.kind.as_str(), "post_swap");
assert!(
q.claim_ready().is_empty(),
"Reconcile must wait for PostSwap, not race it"
);
q.complete_node(id, post_swap.node_id, Ok(()));
let reconcile = claim_one(&q);
assert_eq!(reconcile.kind.as_str(), "reconcile");
q.complete_node(id, reconcile.node_id, Ok(()));
assert_eq!(state_of(&q, id), State::Done);
}
#[test]
fn failed_reconcile_marks_dag_failed() {
let q = JobQueue::new(1);
@ -903,6 +954,7 @@ fn perm_change_shape_prefixes_rebuild_chain() {
"prebuild",
"stop_for_update",
"swap",
"post_swap",
"reconcile",
] {
let c = claim_one(&q);