refactor(#2949): the build-log row carries its node id
`QueueInner` was `{ sched, node_rt }`, where `node_rt` held exactly one
datum per node: the `build_logs` row id. It existed because a `hive_jobq`
node payload is immutable after insert while the log row is created when
the build starts — so the link could not ride the node.
Invert it: the log row names its node (`build_logs.node_id`, one migration
in the existing `schema_versions` framework). Same single-home property,
in the direction the type system allows.
`QueueInner` is now just the scheduler. That is the point: the queue holds
no per-node side map, so nothing has to be locked alongside the graph.
Deleted as a consequence, each surfaced by dead-code analysis after the
edit above rather than predicted:
- `NodeRuntime`, `node_rt`, `set_build_log_id`, and `build_log_id_of`
(which linear-scanned the map to match a wire `u64` against opaque
`NodeId`s). The lookup is an indexed query now.
- `struct Ctx`, entirely. It carried `coord` + `dag_id` + `node_id` into
the executors so the build-log callback could reach the queue; without
the callback, `coord`/`dag_id` were never read and `node_id` was already
on the `Claim` both executors receive.
- `QueueInner::node_running`, which existed only for `set_build_log_id`'s
"only while running" guard.
- The `Fn(i64)` callbacks on `prebuild_toplevel` / `swap_update` /
`priv_run_inner`, replaced by a `node_id: Option<u64>` passed down. The
id travels one way now instead of being registered back.
`meta.rs`'s `nix_logged` passes `None` deliberately: its callers reach it
from outside the queue as well as inside, and nothing reads the link for
them yet.
`id_for_node` takes `MAX(id)` rather than assuming uniqueness — a retried
node opens a second row and the panel wants the current attempt. The test
moved to where the behaviour lives and covers that, plus survival across
completion and non-collision with node-less rows.
This commit is contained in:
atlas
mara
parent
82ef06f445
commit
77cc7bea6b
7 changed files with 150 additions and 162 deletions
|
|
@ -26,25 +26,6 @@ use crate::power::{ReconcileAction, reconcile_action};
|
|||
/// N × this timeout.
|
||||
pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3);
|
||||
|
||||
/// Build-log sink for one claimed node.
|
||||
struct Ctx<'a> {
|
||||
coord: &'a Arc<Coordinator>,
|
||||
dag_id: u64,
|
||||
node_id: super::NodeId,
|
||||
}
|
||||
|
||||
impl Ctx<'_> {
|
||||
fn build_log(&self, log_id: i64) {
|
||||
if self
|
||||
.coord
|
||||
.job_queue
|
||||
.set_build_log_id(self.dag_id, self.node_id, log_id)
|
||||
{
|
||||
self.coord.emit_rebuild_queue_snapshot();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Run one claimed node to completion. Called from a task the
|
||||
/// scheduler spawns per claim; the `Result` (stringified) becomes the
|
||||
/// node's terminal state.
|
||||
|
|
@ -67,19 +48,14 @@ pub(super) async fn run_node(
|
|||
job: super::Job,
|
||||
claim: &Claim,
|
||||
) -> (super::Job, Result<()>) {
|
||||
let ctx = Ctx {
|
||||
coord,
|
||||
dag_id: claim.dag_id,
|
||||
node_id: claim.node_id,
|
||||
};
|
||||
// Every arm is `Result<()>`; the three that grow work declare into `job`
|
||||
// *synchronously*, after their own awaits have finished. Borrowing `&job`
|
||||
// inside an `.await` would make this future non-`Send` (see above), so the
|
||||
// growth executors return what to grow rather than taking the builder.
|
||||
let result = match &claim.kind {
|
||||
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::Prebuild { .. } => run_prebuild(claim).await,
|
||||
NodeKind::Swap { .. } => run_swap(coord, claim).await,
|
||||
NodeKind::PostSwap { .. } => run_post_swap(coord, claim).await,
|
||||
NodeKind::Provision { .. } => run_provision(coord, claim).await,
|
||||
NodeKind::Create { .. } => run_create(claim).await,
|
||||
|
|
@ -241,7 +217,7 @@ async fn run_meta_sync(coord: &Arc<Coordinator>, claim: &Claim, relock: bool) ->
|
|||
/// container is already down: its only purpose is to shrink the swap's
|
||||
/// downtime window, so a stopped agent (no uptime to preserve) doesn't
|
||||
/// pay the double eval — `Swap` builds inline instead.
|
||||
async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result<()> {
|
||||
async fn run_prebuild(claim: &Claim) -> Result<()> {
|
||||
let name = &claim.agent;
|
||||
// Warm the toplevel build only when the container is up — the whole
|
||||
// point of prebuild is to shrink the swap's downtime window. A
|
||||
|
|
@ -249,8 +225,7 @@ async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result<()> {
|
|||
// eval and let the downstream `Swap` build inline.
|
||||
if crate::lifecycle::is_running(name).await {
|
||||
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?;
|
||||
crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(claim.node_id.get())).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -260,17 +235,15 @@ async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result<()> {
|
|||
/// (rev marker, `Rebuilt` event, forge/matrix sync, kick, rescan).
|
||||
/// The recovery-start on failure is NOT here — the DAG's tail
|
||||
/// `Reconcile` runs after this node terminal ok *or* fail.
|
||||
async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<()> {
|
||||
async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
|
||||
let name = &claim.agent;
|
||||
// Swap runs on an already-existing (stopped) container — runtime dir
|
||||
// and listener were created earlier. Pure path accessor suffices.
|
||||
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, &|log_id| {
|
||||
ctx.build_log(log_id);
|
||||
})
|
||||
.await;
|
||||
let result =
|
||||
crate::lifecycle::swap_update(name, &hive, &paths, Some(claim.node_id.get())).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
|
||||
|
|
|
|||
|
|
@ -36,7 +36,6 @@ pub mod templates;
|
|||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Mutex;
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
|
|
@ -101,15 +100,6 @@ pub struct Claim {
|
|||
pub agent: String,
|
||||
}
|
||||
|
||||
/// 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 build-log row link remains host-side (the
|
||||
/// client fetches the log by node id).
|
||||
#[derive(Debug, Default, Clone)]
|
||||
struct NodeRuntime {
|
||||
build_log_id: Option<i64>,
|
||||
}
|
||||
|
||||
/// An owned read-view of a DAG container's carried metadata ([`NodeKind::Dag`]).
|
||||
/// Derived on read from the container node — the data has a single home (the
|
||||
/// node payload); this is not a stored side-table.
|
||||
|
|
@ -119,18 +109,21 @@ struct DagMeta {
|
|||
created_at: DateTime<Utc>,
|
||||
}
|
||||
|
||||
/// The mutable queue state behind the mutex: the crate scheduler plus the
|
||||
/// per-node runtime metadata the graph can't carry. A **DAG is a single
|
||||
/// container node** ([`NodeKind::Dag`], `parent = None`) whose subtree is the
|
||||
/// DAG's work — so the container's `NodeId` is the DAG id, its rolled-up state
|
||||
/// is the DAG state, and there are no grouping side-tables: membership + meta
|
||||
/// are graph queries ([`QueueInner::container`] / [`QueueInner::dag_meta`] +
|
||||
/// the `hive_jobq::Graph` accessors). One shared crate [`Graph`] holds every DAG.
|
||||
/// The mutable queue state behind the mutex: **just the crate scheduler**.
|
||||
/// A **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`)
|
||||
/// whose subtree is the DAG's work — so the container's `NodeId` is the DAG id,
|
||||
/// its rolled-up state is the DAG state, and there are no grouping side-tables:
|
||||
/// membership + meta are graph queries ([`QueueInner::container`] /
|
||||
/// [`QueueInner::dag_meta`] + the `hive_jobq::Graph` accessors). One shared
|
||||
/// crate [`Graph`] holds every DAG.
|
||||
///
|
||||
/// There is deliberately **no per-node side map** any more. The last one held
|
||||
/// the `build_logs` row id; that link now lives on the log row itself
|
||||
/// (`build_logs.node_id`), so it survives a restart and needs no lock held
|
||||
/// alongside the scheduler's — which is what lets the scheduler's own lock be
|
||||
/// the only one the run loop takes.
|
||||
struct QueueInner {
|
||||
sched: Scheduler<NodeKind, Resource>,
|
||||
/// 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>,
|
||||
}
|
||||
|
||||
/// The queue. Lives on `Coordinator` (one per hive-c0re process); a single
|
||||
|
|
@ -208,7 +201,6 @@ impl JobQueue {
|
|||
Self {
|
||||
inner: Mutex::new(QueueInner {
|
||||
sched: Scheduler::new(Graph::new(), table),
|
||||
node_rt: HashMap::new(),
|
||||
}),
|
||||
notify: Notify::new(),
|
||||
}
|
||||
|
|
@ -244,7 +236,6 @@ impl JobQueue {
|
|||
None,
|
||||
)
|
||||
.map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?;
|
||||
inner.node_rt.insert(container, NodeRuntime::default());
|
||||
insert_group(&mut inner, spec.declare, Some(container))?;
|
||||
// Settle the container's own (no-op) logic immediately so it parks in
|
||||
// `Finishing` and its children become runnable — it never needs claiming
|
||||
|
|
@ -374,33 +365,6 @@ impl JobQueue {
|
|||
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();
|
||||
if inner.sched.graph().root_of(node_id).map(NodeId::get) != Some(dag_id)
|
||||
|| !inner.node_running(node_id)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
inner.node_rt.entry(node_id).or_default().build_log_id = Some(log_id);
|
||||
true
|
||||
}
|
||||
|
||||
/// The `build_logs` row id linked to the wire node id `node_id`, if any —
|
||||
/// the lookup behind the `GET /api/build-log/<node_id>` query endpoint (the
|
||||
/// client fetches a node's captured build output on demand rather than
|
||||
/// receiving it inline). Takes the raw wire `u64` (the endpoint's path
|
||||
/// param); `node_rt` is keyed by the opaque `NodeId`, so this scans for the
|
||||
/// matching id — the map is small (live + recently-terminal nodes).
|
||||
#[must_use]
|
||||
pub fn build_log_id_of(&self, node_id: u64) -> Option<i64> {
|
||||
self.lock()
|
||||
.node_rt
|
||||
.iter()
|
||||
.find(|(nid, _)| nid.get() == node_id)
|
||||
.and_then(|(_, rt)| rt.build_log_id)
|
||||
}
|
||||
|
||||
/// The first failed node's error in `dag_id`, if any has failed yet.
|
||||
///
|
||||
/// Unlike the roll-up summary this is readable *mid-flight*, which is the
|
||||
|
|
@ -498,14 +462,6 @@ impl JobQueue {
|
|||
}
|
||||
|
||||
impl QueueInner {
|
||||
/// Whether `id` is a `Running` node.
|
||||
fn node_running(&self, id: NodeId) -> bool {
|
||||
self.sched
|
||||
.graph()
|
||||
.node(id)
|
||||
.is_some_and(|n| n.state == State::Running)
|
||||
}
|
||||
|
||||
/// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals
|
||||
/// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search.
|
||||
fn container(&self, dag_id: u64) -> Option<NodeId> {
|
||||
|
|
@ -593,7 +549,12 @@ impl QueueInner {
|
|||
NodeKind::MetaLock { inputs, .. } => inputs.clone(),
|
||||
_ => Vec::new(),
|
||||
};
|
||||
let build_log_id = self.node_rt.get(&id).and_then(|r| r.build_log_id);
|
||||
// Looked up from the log row itself (`build_logs.node_id`), not a
|
||||
// host-side map. One indexed query per node in the snapshot; the
|
||||
// node set is bounded by `MAX_HISTORY_DAGS` and the store is a
|
||||
// local sqlite file, so this is cheaper than the lock contention
|
||||
// a second shared map would reintroduce.
|
||||
let build_log_id = crate::build_logs::global().and_then(|h| h.id_for_node(id.get()));
|
||||
// `node.parent` is the structural jobq parent. Top-level nodes
|
||||
// have `parent == Some(container)` (direct children of the Dag
|
||||
// container); those become `parent: None` on the wire since the
|
||||
|
|
|
|||
|
|
@ -1544,27 +1544,12 @@ fn deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails() {
|
|||
assert_eq!(state_of(&q, id), State::Failed);
|
||||
}
|
||||
|
||||
// ---- build logs, history ----
|
||||
|
||||
#[test]
|
||||
fn set_build_log_id_links_running_node() {
|
||||
let q = JobQueue::new(1);
|
||||
let id = submit(&q, rebuild("agent-a", "r"));
|
||||
let c = claim_one(&q);
|
||||
assert!(q.set_build_log_id(id, c.node_id, 42));
|
||||
q.complete_node(c.node_id, Ok(()));
|
||||
assert!(
|
||||
!q.set_build_log_id(id, c.node_id, 99),
|
||||
"node no longer running → refused"
|
||||
);
|
||||
// The log id is fetched by node id (the `GET /api/build-log/<id>` lookup),
|
||||
// not carried on the wire — it survives completion in the node runtime.
|
||||
assert_eq!(
|
||||
q.build_log_id_of(c.node_id.get()),
|
||||
Some(42),
|
||||
"log id survives completion"
|
||||
);
|
||||
}
|
||||
// ---- history ----
|
||||
//
|
||||
// The node → build-log link is no longer queue state: the log row carries
|
||||
// `node_id` and the lookup lives in `stores::build_logs` (see
|
||||
// `node_link_survives_completion_and_newest_wins` there). Nothing in the queue
|
||||
// needs testing for it any more, which is the point of that move.
|
||||
|
||||
/// History retention is a **flat** newest-first cap over all terminal DAGs
|
||||
/// (`MAX_HISTORY_DAGS`), not a per-template bucket behind a grace window.
|
||||
|
|
|
|||
Loading…
Reference in a new issue