From 79d4c345bbe5ea3e044a7b55b70eb586a94bd0ed Mon Sep 17 00:00:00 2001 From: damocles Date: Sat, 11 Jul 2026 23:33:45 +0200 Subject: [PATCH] feat(#2349): fan reconcile's start/stop out as first-class dag nodes --- hive-c0re/src/job_queue/exec.rs | 133 +++++++++++++++++---------- hive-c0re/src/job_queue/mod.rs | 34 +++++++ hive-c0re/src/job_queue/model.rs | 19 +++- hive-c0re/src/job_queue/scheduler.rs | 33 +++++-- 4 files changed, 156 insertions(+), 63 deletions(-) diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 66adbe41..89acc641 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -29,6 +29,15 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from pub struct NodeOutput { /// Agents to fan child `Rebuild` DAGs out for (`MetaLock` only). pub fanout: Vec, + /// Mechanical sub-step nodes to append into *this same* DAG at + /// runtime, each depending `AfterOk` on the emitting node — e.g. a + /// `Reconcile` planner emitting a `Start` / `Stop`. A separate + /// channel from `fanout` (which appends whole *child* DAGs) so the + /// sub-step stays a first-class node in the same DAG and the + /// lease-window transient is held across it. The scheduler applies + /// these *before* the emitting node's completion so the DAG never + /// rolls terminal with the appended work still pending. + pub append_nodes: Vec, } /// Step-label + build-log sink for one claimed node. @@ -76,7 +85,9 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< NodeKind::MetaLock { sweep, fanout } => { run_meta_lock(coord, claim, &ctx, *sweep, fanout.clone()).await } - NodeKind::Reconcile => run_reconcile(coord, claim, &ctx).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, @@ -216,6 +227,7 @@ async fn run_meta_lock( } return Ok(NodeOutput { fanout: fanout.unwrap_or_default(), + ..Default::default() }); } let _progress = coord.meta_update_guard(); @@ -230,63 +242,84 @@ async fn run_meta_lock( Some(list) => list, None => meta_update_cascade_agents(&claim.inputs).await, }; - Ok(NodeOutput { fanout: cascade }) + Ok(NodeOutput { + fanout: cascade, + ..Default::default() + }) } -/// Idempotent power converge: `wanted` (durable intent) vs observed. -async fn run_reconcile( - coord: &Arc, - claim: &Claim, - ctx: &Ctx<'_>, -) -> Result { +/// Idempotent power-converge *planner*: compare `wanted` (durable +/// intent) against observed state and, when they diverge, fan the +/// mechanical `Start` / `Stop` out as a first-class node appended to +/// *this* DAG (`NodeOutput::append_nodes`). Does no container work +/// itself — the sub-step becomes visible in the DAG and the +/// lease-window transient (or the sub-step's own node-local guard) +/// rides across it. +async fn run_reconcile(coord: &Arc, claim: &Claim) -> Result { let name = &claim.agent; let running = crate::lifecycle::is_running(name).await; let wanted = coord.power.get_or_seed(name, running)?; - match reconcile_action(wanted, running) { - ReconcileAction::Start => { - // Node-local transient only when the DAG holds none (the - // boot-reconcile template); lease-window guards otherwise - // already cover this node. - let _guard = claim - .transient - .is_none() - .then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Starting)); - // Run the typed start preamble: ensures the runtime dir - // exists and writes the nspawn/resource-limits drop-ins. - // The returned StartableAgent token is the only way to call - // start_with_fallback — omitting this becomes a compile error. - let agent_dir = crate::paths::agent_runtime_dir(name); - 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 up and will connect to this socket on its first turn. - // Event-driven (no background poll) — c0re owns the listener - // lifecycle, so register here rather than waiting for a sweep. - coord.register_agent(name)?; - coord.kick_agent(name, "container started"); - coord.rescan_containers_and_emit().await; - } - ReconcileAction::Stop => { - 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 { - agent: name.clone(), - }); - coord.rescan_containers_and_emit().await; - } + let append_nodes = match reconcile_action(wanted, running) { + ReconcileAction::Start => vec![NodeKind::Start], + ReconcileAction::Stop => vec![NodeKind::Stop], ReconcileAction::Noop => { tracing::debug!(%name, wanted = wanted.as_str(), running, "reconcile: noop"); + Vec::new() } - } + }; + Ok(NodeOutput { + append_nodes, + ..Default::default() + }) +} + +/// 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, claim: &Claim, ctx: &Ctx<'_>) -> Result { + 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 + // transient already covers this node. + let _guard = claim + .transient + .is_none() + .then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Starting)); + // Run the typed start preamble: ensures the runtime dir exists and + // writes the nspawn/resource-limits drop-ins. The returned + // StartableAgent token is the only way to call start_with_fallback — + // omitting this becomes a compile error. + let agent_dir = crate::paths::agent_runtime_dir(name); + 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 + // up and will connect to this socket on its first turn. Event-driven + // (no background poll) — c0re owns the listener lifecycle, so + // register here rather than waiting for a sweep. + coord.register_agent(name)?; + coord.kick_agent(name, "container started"); + coord.rescan_containers_and_emit().await; + Ok(NodeOutput::default()) +} + +/// 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, claim: &Claim, ctx: &Ctx<'_>) -> Result { + 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 { + agent: name.clone(), + }); + coord.rescan_containers_and_emit().await; Ok(NodeOutput::default()) } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index bfb118b2..27a968ee 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -169,6 +169,40 @@ impl JobQueue { ids } + /// Append a node into a *live* (non-terminal) DAG at runtime, + /// depending `AfterOk` on `dep_on` (the node that emitted it). Lets a + /// planner node — e.g. [`NodeKind::Reconcile`] — fan a mechanical + /// sub-step ([`NodeKind::Start`] / [`NodeKind::Stop`]) out as a + /// first-class node in the *same* DAG. + /// + /// Must be called *before* the emitting node's [`Self::complete_node`] + /// so the DAG doesn't roll terminal with the new node still pending — + /// that keeps the lease-window transient held across the sub-step and + /// lets the appended node's `AfterOk` dep resolve as soon as the + /// emitter settles `Done`. No-op (returns `None`) if the DAG is gone. + pub fn append_node(&self, dag_id: u64, kind: NodeKind, dep_on: NodeId) -> Option { + let mut inner = self.inner.lock().expect("job_queue mutex poisoned"); + let dag = inner.dags.iter_mut().find(|d| d.id == dag_id)?; + let new_id: NodeId = u32::try_from(dag.nodes.len()).unwrap_or(u32::MAX); + dag.nodes.push(Node { + id: new_id, + kind, + deps: vec![model::Dep { + on: dep_on, + when: DepWhen::AfterOk, + }], + state: State::Queued, + step: None, + build_log_id: None, + started_at: None, + finished_at: None, + error: None, + }); + drop(inner); + self.notify.notify_one(); + Some(new_id) + } + fn dedup_target<'a>(inner: &'a mut Inner, spec: &DagSpec) -> Option<&'a mut Dag> { inner.dags.iter_mut().find(|d| { d.rollup() == State::Queued diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index c572e7ae..98026947 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -90,10 +90,21 @@ pub enum NodeKind { sweep: bool, fanout: Option>, }, - /// Idempotent power converge: read `wanted` + observed state; - /// start if `Up` & down (with cold-start fallback), stop if - /// `Offline` & up, else noop. + /// Idempotent power converge *planner*: read `wanted` + observed + /// state and decide the action (start if `Up` & down, stop if + /// `Offline` & up, else noop). The mechanical work is not done in + /// this node — it fans a child [`NodeKind::Start`] / [`NodeKind::Stop`] + /// DAG out at runtime so the sub-step is a first-class DAG node. Reconcile, + /// Mechanical container start: the start preamble (runtime dir + + /// drop-ins), `start_with_fallback`, MCP listener registration, and + /// the manager kick. Fanned out by a [`NodeKind::Reconcile`] that + /// observed `wanted = Up` and the container down. + Start, + /// Mechanical container stop: `nixos-container` kill, MCP listener + /// unregister, and the `Killed` manager notify. Fanned out by a + /// [`NodeKind::Reconcile`] that observed `wanted = Offline` and up. + Stop, /// Mechanical `nixos-container stop` for the profile swap. Never /// touches `wanted`. Noop if already stopped. StopForUpdate, @@ -125,6 +136,8 @@ impl NodeKind { NodeKind::Create => "create", NodeKind::MetaLock { .. } => "meta_lock", NodeKind::Reconcile => "reconcile", + NodeKind::Start => "start", + NodeKind::Stop => "stop", NodeKind::StopForUpdate => "stop_for_update", NodeKind::Signal => "signal", NodeKind::Drain => "drain", diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index fb7997a4..230f9d30 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -92,14 +92,32 @@ async fn handle_completion( done: NodeDone, ) { let NodeDone { claim, result } = done; - let (queue_result, fanout) = match result { + match result { Ok(output) => { tracing::info!( dag = claim.dag_id, node = claim.node_id, "job_queue: node done" ); - (Ok(()), output.fanout) + // Append any in-DAG sub-step nodes (e.g. a `Reconcile` + // planner's `Start` / `Stop`) BEFORE completing this node, so + // completing it doesn't roll the DAG terminal while the + // appended work is still pending — that keeps the lease-window + // transient held across the sub-step. Each depends `AfterOk` + // on this node, so it becomes ready the instant this one + // settles `Done` just below. + for kind in output.append_nodes { + coord + .job_queue + .append_node(claim.dag_id, kind, claim.node_id); + } + coord + .job_queue + .complete_node(claim.dag_id, claim.node_id, Ok(())); + if !output.fanout.is_empty() { + let specs = fanout_specs(&claim, output.fanout); + coord.job_queue.append_children(specs); + } } Err(e) => { let msg = format!("{e:#}"); @@ -111,15 +129,10 @@ async fn handle_completion( error = %msg, "job_queue: node failed" ); - (Err(msg), Vec::new()) + coord + .job_queue + .complete_node(claim.dag_id, claim.node_id, Err(msg)); } - }; - coord - .job_queue - .complete_node(claim.dag_id, claim.node_id, queue_result); - if !fanout.is_empty() { - let specs = fanout_specs(&claim, fanout); - coord.job_queue.append_children(specs); } process_terminals(coord, transients).await; coord.emit_rebuild_queue_snapshot();