884e39ba63
mara on !2910: "why is set_transient still a thing if it completely derives from nodes?" It was still a thing because the scheduler mirrored the derived set into a stored map that every consumer read — derived state computed once and then cached, with the reconciliation loop existing only to keep the cache honest. `transient_snapshot()` now derives: `running_transients()` off the live graph, with the handful of entries that have no node behind them (destroy, migration) overlaid on top. There is no cached copy left to go stale or disagree with what is running. `set_transient` / `clear_transient` split by what they actually do: `set_manual_transient` / `clear_manual_transient` own the stored map for the no-node callers, and `emit_transient_set` / `emit_transient_cleared` publish the edges both paths need. Two things had to survive, and both are edges rather than state: - The dashboard's `TransientSet` / `TransientCleared` events. The scheduler carries the previous derived value and emits the diff. - The crash watcher's grace window. `recent_transient_within` answers "was a transient cleared just now?", which is what stops a deliberate stop from reading as a crash on the next 10s poll — a derived read of current state cannot answer it, so the clear still stamps. The scheduler keeps `deliberate_stop` alongside the label precisely so it is available at clear time: the node it came from is, by definition, no longer running to be asked. `TransientState::since` becomes wall-clock and, for derived entries, is the node's own `started_at` — the true start of the operation rather than the moment a watcher first noticed it, which is what the old guard-creation timestamp actually measured. `running_transients` returns a named `RunningTransient` rather than a 4-tuple; two of its fields are strings and one is a bool whose meaning is not guessable at a call site. Note for anyone reaching for a timestamp here: chrono is vendored with `default-features = false`, so there is no `Utc::now()`. The workspace convention is `wire_time::now_unix()` / `from_secs()`. Checked with clippy (`--all-targets -D warnings`), `cargo test -p hive-c0re -p hive-jobq` (322 + 41 passed) and `nix fmt`.
702 lines
30 KiB
Rust
702 lines
30 KiB
Rust
//! Generic job-DAG queue + desired-state reconciliation — the host-side
|
|
//! wrapper over the domain-agnostic [`hive_jobq`] scheduler. Jobs are nodes in
|
|
//! per-request DAGs (see [`templates`]); the special cases (graceful-stop
|
|
//! watcher, deferred-start follow-up, meta-update cascade) collapse into DAG
|
|
//! *shapes* over the shared node primitives ([`model::NodeKind`]).
|
|
//!
|
|
//! [`hive_jobq`] owns the graph, the two-class resource pool, and the roll-up
|
|
//! settle loop; this module maps hive-c0re's concepts onto it:
|
|
//! - [`model::NodeKind`] **is** the crate payload `N` directly — each variant
|
|
//! carries the agent it targets ([`NodeKind::agent`]); the two resource
|
|
//! classes are [`resource::Resource`] (`BuildSlot` node-held, `Agent` lease
|
|
//! subtree-held), derived per node by [`NodeKind::resource_deps`];
|
|
//! - a **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`)
|
|
//! carrying the group's metadata, with the work nodes hung under it as
|
|
//! its subtree (the **parent axis** groups; `deps` order). So the container's
|
|
//! `NodeId` is the DAG id, its rolled-up state is the DAG state, and membership
|
|
//! is a graph walk — there are no host grouping side-tables. The lease is owned
|
|
//! by a subtree root and borrowed by its descendants (continuity);
|
|
//! - per-DAG terminal work is an ordinary **tail node**
|
|
//! ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]) that the builder
|
|
//! appends in [`templates`], edged onto the DAG's other group roots by the
|
|
//! outcome it reports. Templates emit one tail per outcome and the graph runs
|
|
//! exactly one, so nothing branches at runtime.
|
|
//!
|
|
//! The queue is runtime-only (no persistence): an empty graph on boot; desired
|
|
//! state is re-derived by the reconcile sweep. A single scheduler task
|
|
//! ([`scheduler::run_worker`]) drives it; concurrency comes from the build-slot
|
|
//! capacity, not multiple workers. Design: `docs/coordinator.md::Job queue`.
|
|
|
|
pub mod exec;
|
|
pub mod model;
|
|
pub mod resource;
|
|
pub mod scheduler;
|
|
pub mod submit;
|
|
pub mod templates;
|
|
#[cfg(test)]
|
|
mod tests;
|
|
|
|
use std::collections::HashMap;
|
|
use std::sync::Mutex;
|
|
|
|
use chrono::{DateTime, Utc};
|
|
use hive_host_sock::jobs::NodeView;
|
|
use hive_jobq::resources::ResourceTable;
|
|
use hive_jobq::scheduler::{Outcome, Scheduler};
|
|
use hive_jobq::{Dep, Graph, NodeId};
|
|
use hive_sh4re::wire_time::now_unix;
|
|
use tokio::sync::Notify;
|
|
|
|
pub use hive_jobq::TerminalState;
|
|
pub use model::{DagSpec, DagView, NodeKind, NodeSpec, PermPayload, Source, State};
|
|
use resource::Resource;
|
|
|
|
/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) the snapshot
|
|
/// retains, newest first. A flat cap over the whole sorted list: the
|
|
/// dashboard renders one recent-builds list, so one number bounds it.
|
|
const MAX_HISTORY_DAGS: usize = 50;
|
|
|
|
/// Cap on stored node error strings.
|
|
const MAX_ERROR_LEN: usize = 2_000;
|
|
|
|
/// One live transient pill, derived from a running node.
|
|
///
|
|
/// A named struct rather than a tuple because three of its four fields are
|
|
/// easy to confuse at a call site: two are strings and two answer questions
|
|
/// nobody should have to guess at ("is this the agent or the label?", "does
|
|
/// this bool mean deliberate or running?").
|
|
#[derive(Debug, Clone)]
|
|
pub struct RunningTransient {
|
|
/// The agent whose lease the node declared.
|
|
pub agent: String,
|
|
/// The node's own wire tag, rendered as the pill.
|
|
pub label: String,
|
|
/// Whether this operation is expected to take the container down — the
|
|
/// crash watcher's input. See [`NodeKind::takes_container_down`].
|
|
pub takes_container_down: bool,
|
|
/// When the node started running, so the dashboard can tick elapsed
|
|
/// seconds. Taken from the node itself, which is the true start of the
|
|
/// operation rather than the moment a watcher noticed it.
|
|
pub since: DateTime<Utc>,
|
|
}
|
|
|
|
/// A node claimed for execution — everything the executor needs, snapshotted at
|
|
/// claim time.
|
|
#[derive(Debug, Clone)]
|
|
pub struct Claim {
|
|
pub dag_id: u64,
|
|
pub node_id: NodeId,
|
|
pub kind: NodeKind,
|
|
/// The agent this node targets (its own, not a DAG-level field). Empty for
|
|
/// the agentless [`NodeKind::MetaLock`] + [`NodeKind::Dag`] container nodes.
|
|
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.
|
|
struct DagMeta {
|
|
source: Source,
|
|
reason: String,
|
|
created_at: i64,
|
|
}
|
|
|
|
/// 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.
|
|
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
|
|
/// scheduler task ([`scheduler::run_worker`]) drives it.
|
|
pub struct JobQueue {
|
|
inner: Mutex<QueueInner>,
|
|
/// Wakes the scheduler when something new arrives or state changed.
|
|
pub(crate) notify: Notify,
|
|
}
|
|
|
|
impl std::fmt::Debug for JobQueue {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("JobQueue").finish_non_exhaustive()
|
|
}
|
|
}
|
|
|
|
impl Default for JobQueue {
|
|
fn default() -> Self {
|
|
Self::new(1)
|
|
}
|
|
}
|
|
|
|
/// Insert `nodes` into the shared graph, honouring the spec's explicit **parent
|
|
/// axis**: a node with `parent = None` is a top-level group root (re-parented to
|
|
/// `group_parent`, which is `None` for `submit` and the emitting node for
|
|
/// `append_subgraph`); a node with `parent = Some(idx)` becomes a child of the
|
|
/// already-inserted node at spec index `idx`. `deps` are translated to crate
|
|
/// `Dep::Node` edges verbatim — templates declare the parent axis + sibling
|
|
/// ordering directly, so there is no dep-on-root to drop and no lease to hoist:
|
|
/// each node declares its own `Dep::Resource`, and the crate's borrow model
|
|
/// keeps a resource continuous across a subtree (a root owns it, descendants
|
|
/// borrow it). Independent group roots (multiple `parent = None` nodes) carry no
|
|
/// cross-links, so a multi-agent DAG's per-agent subgraphs run concurrently, each
|
|
/// on its own lease. Records per-node `node_rt`. Returns the inserted ids
|
|
/// (index-aligned with `nodes`). A node with `parent = None` is re-parented to
|
|
/// `group_parent` (the DAG container for a template, or the emitting node for a
|
|
/// runtime-appended subgraph); a node's `parent` / dep targets must precede it
|
|
/// in `nodes` (submit-time `validate` enforces density + acyclicity).
|
|
///
|
|
/// # Errors
|
|
/// Propagates a crate graph-insert error (malformed dep/parent / dep-scope).
|
|
fn insert_group(
|
|
inner: &mut QueueInner,
|
|
nodes: &[NodeSpec],
|
|
group_parent: Option<NodeId>,
|
|
) -> anyhow::Result<Vec<NodeId>> {
|
|
let mut ids: Vec<NodeId> = Vec::with_capacity(nodes.len());
|
|
for ns in nodes {
|
|
let payload = ns.kind.clone();
|
|
let mut deps = payload.resource_deps();
|
|
for d in &ns.deps {
|
|
deps.push(Dep::Node {
|
|
id: ids[dep_index(d.on)],
|
|
when: d.when,
|
|
});
|
|
}
|
|
let parent = match ns.parent {
|
|
Some(idx) => Some(ids[dep_index(idx)]),
|
|
None => group_parent,
|
|
};
|
|
let id = inner
|
|
.sched
|
|
.append(payload, deps, parent)
|
|
.map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
|
|
ids.push(id);
|
|
inner.node_rt.insert(id, NodeRuntime::default());
|
|
}
|
|
Ok(ids)
|
|
}
|
|
|
|
impl JobQueue {
|
|
#[must_use]
|
|
pub fn new(build_slots: usize) -> Self {
|
|
let mut table = ResourceTable::new();
|
|
table.set_capacity(
|
|
Resource::BuildSlot,
|
|
u32::try_from(build_slots.max(1)).unwrap_or(u32::MAX),
|
|
);
|
|
Self {
|
|
inner: Mutex::new(QueueInner {
|
|
sched: Scheduler::new(Graph::new(), table),
|
|
node_rt: HashMap::new(),
|
|
}),
|
|
notify: Notify::new(),
|
|
}
|
|
}
|
|
|
|
fn lock(&self) -> std::sync::MutexGuard<'_, QueueInner> {
|
|
self.inner.lock().expect("job_queue mutex poisoned")
|
|
}
|
|
|
|
/// Submit a DAG. Validates the spec, inserts a [`NodeKind::Dag`] **container
|
|
/// node** carrying the group's metadata, then inserts the template's nodes as
|
|
/// its subtree (their roots re-parented to the container). Returns the
|
|
/// container's id as the DAG id — its rolled-up state is the DAG state.
|
|
///
|
|
/// # Errors
|
|
/// Propagates the spec-validation error (empty / cyclic / bad parent) or a
|
|
/// graph-insert error (dependencies that aren't dependency-topological).
|
|
pub fn submit(&self, spec: DagSpec) -> anyhow::Result<u64> {
|
|
templates::validate(&spec)?;
|
|
let mut inner = self.lock();
|
|
let container = inner
|
|
.sched
|
|
.append(
|
|
NodeKind::Dag {
|
|
source: spec.source,
|
|
reason: spec.reason,
|
|
created_at: now_unix(),
|
|
},
|
|
Vec::new(),
|
|
None,
|
|
)
|
|
.map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?;
|
|
inner.node_rt.insert(container, NodeRuntime::default());
|
|
insert_group(&mut inner, &spec.nodes, 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
|
|
// or executing, and stays out of `claim_ready`. It rolls up terminal when
|
|
// its whole subtree settles (that's the DAG-done signal).
|
|
inner.sched.complete(container, Outcome::Done);
|
|
drop(inner);
|
|
self.notify.notify_one();
|
|
Ok(container.get())
|
|
}
|
|
|
|
/// Append a whole *subgraph* into a live DAG at runtime — the single
|
|
/// in-DAG-growth primitive. The subgraph is inserted as a [`insert_group`]
|
|
/// rooted under `dep_on` (the emitting node): the subgraph's own root becomes
|
|
/// a *child* of `dep_on`, its steps children of that root, and the group's
|
|
/// agent lease is hoisted onto that root. Ordering root→`dep_on` is the parent
|
|
/// gate — the children run once `dep_on` reaches `Finishing`. Because the
|
|
/// emitting node stays `Finishing` until this appended subtree is terminal and
|
|
/// the DAG's terminal node deps on the top root, roll-up keeps the DAG from
|
|
/// settling early with no explicit wiring. Returns the new node ids; empty if
|
|
/// the DAG is gone or `nodes` is empty.
|
|
pub fn append_subgraph(&self, dag_id: u64, nodes: &[NodeSpec], dep_on: NodeId) -> Vec<NodeId> {
|
|
if nodes.is_empty() {
|
|
return Vec::new();
|
|
}
|
|
let mut inner = self.lock();
|
|
if inner.container(dag_id).is_none() {
|
|
return Vec::new();
|
|
}
|
|
// Insert the subgraph as a group rooted under the emitting node: the
|
|
// subgraph's own root becomes a child of `dep_on`, its steps children of
|
|
// that root. No terminal-node wiring — roll-up carries terminality: the
|
|
// emitter stays `Finishing` until this appended subtree settles, and the
|
|
// container node rolls up terminal only once its whole subtree (incl. this
|
|
// appended work) has settled, so the DAG hook waits for free.
|
|
let ids = match insert_group(&mut inner, nodes, Some(dep_on)) {
|
|
Ok(ids) => ids,
|
|
Err(e) => {
|
|
tracing::error!(
|
|
dag = dag_id,
|
|
error = %e,
|
|
"job_queue: append_subgraph insert failed"
|
|
);
|
|
return Vec::new();
|
|
}
|
|
};
|
|
drop(inner);
|
|
self.notify.notify_one();
|
|
ids
|
|
}
|
|
|
|
/// Claim every currently-runnable node, acquiring its resources, and mark it
|
|
/// `Running`. Delegates readiness + resource acquisition to the crate's
|
|
/// settle loop; builds a [`Claim`] per started node from its payload + its
|
|
/// DAG container's metadata. The container node itself is claimed like any
|
|
/// other (its executor is an instant no-op that lets its subtree start).
|
|
pub fn claim_ready(&self) -> Vec<Claim> {
|
|
let mut inner = self.lock();
|
|
let inner = &mut *inner;
|
|
let started = inner.sched.settle();
|
|
let mut claims = Vec::with_capacity(started.len());
|
|
for id in started {
|
|
let Some(node) = inner.sched.graph().node(id) else {
|
|
continue;
|
|
};
|
|
let kind = node.payload.clone();
|
|
let agent = node.payload.agent().to_owned();
|
|
let Some(container) = inner.sched.graph().root_of(id) else {
|
|
continue;
|
|
};
|
|
claims.push(Claim {
|
|
dag_id: container.get(),
|
|
node_id: id,
|
|
kind,
|
|
agent,
|
|
});
|
|
// `started_at` is stamped on the graph `Node` by the scheduler's
|
|
// transition to `Running` — no host-side copy needed.
|
|
}
|
|
claims
|
|
}
|
|
|
|
/// Mark a claimed node terminal, recording its outcome + (truncated) error.
|
|
/// The crate releases the node's build slot immediately and cascades the
|
|
/// `AfterOk` failure cancellation + subtree lease release.
|
|
///
|
|
/// Nothing is returned: a DAG's terminal side effects are its own tail nodes
|
|
/// ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]), which the
|
|
/// scheduler claims and runs like any other node.
|
|
pub fn complete_node(&self, node_id: NodeId, result: Result<(), String>) {
|
|
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, so there is nothing to clear here.
|
|
let outcome = match result {
|
|
Ok(()) => Outcome::Done,
|
|
Err(e) => Outcome::Failed(truncate_error(&e)),
|
|
};
|
|
inner.sched.complete(node_id, outcome);
|
|
drop(inner);
|
|
self.notify.notify_one();
|
|
}
|
|
|
|
/// Cancel a DAG that hasn't started yet: every work node is still `Pending`,
|
|
/// so each is cancelled. `false` once any work node is running or terminal —
|
|
/// an in-flight nix build isn't interruptible.
|
|
///
|
|
/// **Nodes that explicitly observe cancellation are spared** — a node whose
|
|
/// edge names [`hive_jobq::TerminalState::Cancelled`] is asking to run when
|
|
/// the work it follows was dropped, which is exactly what an approval tail
|
|
/// needs: cancel the work, and the tail still fires to resolve the approval
|
|
/// row rather than leaving it dangling forever.
|
|
///
|
|
/// Nothing is special-cased by node kind. `AFTER_ANY` deliberately does *not*
|
|
/// accept `Cancelled`, so an ordinary weak-edged step (rebuild's `Reconcile`,
|
|
/// say) is cancelled along with everything else — there is nothing to converge
|
|
/// when no node ever ran. Only a node that named `Cancelled` survives, and it
|
|
/// survives because it asked to.
|
|
///
|
|
/// `id` names **any node**, not specifically a DAG. Cancelling a group root
|
|
/// drops that whole group (the cascade is the scheduler's), which is what
|
|
/// the dashboard's whole-DAG cancel does; cancelling an interior node drops
|
|
/// just that branch. Nothing here knows about DAGs.
|
|
pub fn cancel(&self, id: u64) -> bool {
|
|
let mut inner = self.lock();
|
|
let Some(node) = inner.sched.graph().resolve_id(id) else {
|
|
return false;
|
|
};
|
|
if !inner.sched.cancel_node(node) {
|
|
return false;
|
|
}
|
|
drop(inner);
|
|
self.notify.notify_one();
|
|
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
|
|
/// point: a compensation node runs `AfterAny` its subject, so when it asks,
|
|
/// the DAG is still `Finishing` (the compensation node itself is running)
|
|
/// while the node it is compensating for has already settled `Failed`. That
|
|
/// lets the compensation annotate its bookkeeping with the reason the deploy
|
|
/// failed, instead of having the error handed down from the node that hit
|
|
/// it. `None` when nothing has failed — the ordinary success path.
|
|
#[must_use]
|
|
pub fn first_error(&self, dag_id: u64) -> Option<String> {
|
|
let inner = self.lock();
|
|
let container = inner.container(dag_id)?;
|
|
inner
|
|
.sched
|
|
.graph()
|
|
.first_error(container)
|
|
.map(ToOwned::to_owned)
|
|
}
|
|
|
|
/// `(agent, label, takes_container_down)` for the live transient-pill set,
|
|
/// recomputed from the nodes **actually running** — not from an intent a
|
|
/// template declared at submit time. (A rebuild used to report `rebuilding`
|
|
/// for its whole life: prebuild, stop, swap, tail and reconcile alike.)
|
|
///
|
|
/// A node lights a pill when it is `Running` **and declares the agent's
|
|
/// resource itself**. Declaring is the test, not targeting — `Prebuild` /
|
|
/// `MetaSync` name an agent but are lease-exempt on purpose, since the
|
|
/// container keeps serving through them. Nor is it the lease *owner*:
|
|
/// `resource_state()` answers "who holds the slot", a different question.
|
|
///
|
|
/// `label` is the node's own wire tag ([`NodeKind::as_str`]), the vocabulary
|
|
/// [`NodeView::kind`] already ships, so a pill and a DAG node name an
|
|
/// operation identically. `takes_container_down` is the crash watcher's
|
|
/// input, carried rather than inferred from the label — a `Start` pill and a
|
|
/// `Stop` pill are both pills; only one means a vanished container is
|
|
/// expected.
|
|
///
|
|
/// By design, `Start` / `Stop` / `PostSwap` run inside a lease-holding
|
|
/// ancestor and re-declare nothing, so they light no pill; closing that is
|
|
/// the resources-where-constructed work, not this function. An agent's lease
|
|
/// is cap-1, so at most one entry per agent.
|
|
#[must_use]
|
|
pub fn running_transients(&self) -> Vec<RunningTransient> {
|
|
let inner = self.lock();
|
|
inner
|
|
.sched
|
|
.graph()
|
|
.nodes()
|
|
.filter(|n| matches!(n.state, State::Running))
|
|
.filter_map(|n| {
|
|
let agent = n
|
|
.payload
|
|
.resource_deps()
|
|
.into_iter()
|
|
.find_map(|d| match d {
|
|
Dep::Resource {
|
|
name: Resource::Agent(a),
|
|
..
|
|
} => Some(a),
|
|
_ => None,
|
|
})?;
|
|
Some(RunningTransient {
|
|
agent,
|
|
label: n.payload.as_str().to_owned(),
|
|
takes_container_down: n.payload.takes_container_down(),
|
|
// `started_at` is set when a node enters `Running`, and this
|
|
// only sees `Running` nodes — the fallback is unreachable in
|
|
// practice, and "just now" is the honest answer if it isn't.
|
|
since: n
|
|
.started_at
|
|
.unwrap_or_else(|| hive_sh4re::wire_time::from_secs(now_unix())),
|
|
})
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`.
|
|
#[must_use]
|
|
pub fn snapshot(&self) -> Vec<DagView> {
|
|
let inner = self.lock();
|
|
let mut ids = inner.visible_dags();
|
|
ids.sort_unstable_by_key(|c| c.get());
|
|
ids.into_iter().filter_map(|c| inner.dag_view(c)).collect()
|
|
}
|
|
|
|
/// Number of live (non-terminal) DAGs — tests + diagnostics.
|
|
#[cfg(test)]
|
|
#[must_use]
|
|
pub fn live_count(&self) -> usize {
|
|
let inner = self.lock();
|
|
inner
|
|
.containers()
|
|
.into_iter()
|
|
.filter(|&c| inner.sched.graph().is_settled(c) == Some(false))
|
|
.count()
|
|
}
|
|
}
|
|
|
|
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> {
|
|
self.sched.graph().nodes().find_map(|n| {
|
|
(n.parent.is_none()
|
|
&& n.id.get() == dag_id
|
|
&& matches!(n.payload, NodeKind::Dag { .. }))
|
|
.then_some(n.id)
|
|
})
|
|
}
|
|
|
|
/// The container's carried domain metadata as an owned read-view. The data
|
|
/// lives solely in the [`NodeKind::Dag`] payload — this is a derived read,
|
|
/// not a stored side-table.
|
|
fn dag_meta(&self, container: NodeId) -> Option<DagMeta> {
|
|
let NodeKind::Dag {
|
|
source,
|
|
reason,
|
|
created_at,
|
|
} = &self.sched.graph().node(container)?.payload
|
|
else {
|
|
return None;
|
|
};
|
|
Some(DagMeta {
|
|
source: *source,
|
|
reason: reason.clone(),
|
|
created_at: *created_at,
|
|
})
|
|
}
|
|
|
|
/// Project a DAG into its wire [`DagView`]: a near-raw view of the
|
|
/// container's work nodes, with `Done` nodes excluded. Lifecycle
|
|
/// (`state` / `started_at` / `finished_at` / `error`) is read straight
|
|
/// off each `hive_jobq::Node`; the client derives the DAG label, roll-up
|
|
/// state, and DAG timestamps from the node set. Non-derivable per-node
|
|
/// payload (`approval_id`, meta `inputs`) rides the owning node. Returns
|
|
/// `None` when every work node is `Done` or `Skipped` — a fully-settled
|
|
/// DAG drops out of the snapshot entirely (a `Failed` one lingers until
|
|
/// aged out).
|
|
fn dag_view(&self, container: NodeId) -> Option<DagView> {
|
|
let meta = self.dag_meta(container)?;
|
|
let mut nodes = Vec::new();
|
|
// Whether anything in this DAG still has an outcome worth showing.
|
|
// Kept separate from `nodes` being non-empty: skipped nodes ride the
|
|
// wire so the dashboard can mark the branches that weren't taken, but
|
|
// they must not by themselves hold a finished DAG in the snapshot.
|
|
let mut any_unsettled = false;
|
|
// DAG-level timestamps are taken over *all* subtree nodes (including the
|
|
// `Done` ones excluded from the wire) — the client can't derive them
|
|
// from a `Done`-filtered node set, so the host computes them here.
|
|
let mut started: Vec<DateTime<Utc>> = Vec::new();
|
|
let mut finished: Vec<DateTime<Utc>> = Vec::new();
|
|
for node in self.sched.graph().descendants(container) {
|
|
let id = node.id;
|
|
if let Some(s) = node.started_at {
|
|
started.push(s);
|
|
}
|
|
if let Some(f) = node.finished_at {
|
|
finished.push(f);
|
|
}
|
|
// `Done` nodes drop off the wire — a finished step isn't
|
|
// interesting. `Skipped` ones stay: which branch a run *didn't*
|
|
// take is the readable half of an outcome-branched DAG.
|
|
if matches!(node.state, State::Done) {
|
|
continue;
|
|
}
|
|
any_unsettled |= !matches!(node.state, State::Skipped);
|
|
let deps: Vec<u64> = node
|
|
.deps
|
|
.iter()
|
|
.filter_map(|d| match d {
|
|
Dep::Node { id, .. } => Some(id.get()),
|
|
Dep::Resource { .. } => None,
|
|
})
|
|
.collect();
|
|
// Non-derivable per-node payload rides the node that owns it. Every
|
|
// deploy phase carries the approval id, but only the subtree root
|
|
// projects it onto the wire — hanging the approval link off all of
|
|
// them would render the same card once per phase.
|
|
let approval_id = match &node.payload {
|
|
NodeKind::DeployWindow { approval_id, .. } => Some(*approval_id),
|
|
_ => None,
|
|
};
|
|
let inputs = match &node.payload {
|
|
NodeKind::MetaLock { inputs, .. } => inputs.clone(),
|
|
_ => Vec::new(),
|
|
};
|
|
let build_log_id = self.node_rt.get(&id).and_then(|r| r.build_log_id);
|
|
// `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
|
|
// container itself is not part of the work-node payload. Sub-nodes
|
|
// carry the id of their containing parent work-node.
|
|
let parent = node
|
|
.parent
|
|
.filter(|&p| p != container)
|
|
.map(hive_jobq::NodeId::get);
|
|
nodes.push(NodeView {
|
|
id: id.get(),
|
|
agent: node.payload.agent().to_owned(),
|
|
kind: node.payload.as_str().to_owned(),
|
|
deps,
|
|
state: node.state,
|
|
started_at: node.started_at,
|
|
finished_at: node.finished_at,
|
|
error: node.error.clone(),
|
|
approval_id,
|
|
inputs,
|
|
build_log_id,
|
|
parent,
|
|
});
|
|
}
|
|
if !any_unsettled {
|
|
return None;
|
|
}
|
|
let is_terminal = self.sched.graph().is_settled(container) == Some(true);
|
|
Some(DagView {
|
|
id: container.get(),
|
|
source: meta.source,
|
|
reason: meta.reason.clone(),
|
|
created_at: hive_sh4re::wire_time::from_secs(meta.created_at),
|
|
started_at: started.into_iter().min(),
|
|
finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(),
|
|
nodes,
|
|
})
|
|
}
|
|
|
|
/// When a DAG's work node finishes on `finished_at` — the max over its
|
|
/// subtree (read off the graph `Node`, as unix seconds), for the history
|
|
/// cap ordering.
|
|
fn dag_finished_at(&self, container: NodeId) -> i64 {
|
|
self.sched
|
|
.graph()
|
|
.descendants(container)
|
|
.filter_map(|n| n.finished_at)
|
|
.map(|t| t.timestamp())
|
|
.max()
|
|
.unwrap_or(0)
|
|
}
|
|
|
|
/// Every DAG container node id in the graph.
|
|
fn containers(&self) -> Vec<NodeId> {
|
|
self.sched
|
|
.graph()
|
|
.nodes()
|
|
.filter(|n| n.parent.is_none() && matches!(n.payload, NodeKind::Dag { .. }))
|
|
.map(|n| n.id)
|
|
.collect()
|
|
}
|
|
|
|
/// The **visible** DAG set for the snapshot: every live (non-terminal) DAG,
|
|
/// plus the newest [`MAX_HISTORY_DAGS`] terminal ones. Crate nodes for
|
|
/// evicted DAGs linger in the graph (bounded-prune is a Stage-C follow-up);
|
|
/// this filter is what bounds what the dashboard sees.
|
|
fn visible_dags(&self) -> Vec<NodeId> {
|
|
let mut live: Vec<NodeId> = Vec::new();
|
|
let mut terminal: Vec<(NodeId, i64)> = Vec::new();
|
|
for c in self.containers() {
|
|
if self.sched.graph().is_settled(c) == Some(true) {
|
|
terminal.push((c, self.dag_finished_at(c)));
|
|
} else {
|
|
live.push(c);
|
|
}
|
|
}
|
|
// Newest first, so truncating to the cap keeps the most recent.
|
|
terminal.sort_by(|a, b| b.1.cmp(&a.1).then(b.0.get().cmp(&a.0.get())));
|
|
terminal.truncate(MAX_HISTORY_DAGS);
|
|
let mut kept = live;
|
|
kept.extend(terminal.into_iter().map(|(c, _)| c));
|
|
kept
|
|
}
|
|
}
|
|
|
|
/// A spec dependency index (`Dep.on`, a wire `u64`) as a `usize` for indexing
|
|
/// into the node/id vectors. `templates::validate` guarantees it's in range.
|
|
fn dep_index(on: u64) -> usize {
|
|
usize::try_from(on).unwrap_or(usize::MAX)
|
|
}
|
|
|
|
/// Truncate a node error to [`MAX_ERROR_LEN`] on a char boundary, appending `…`.
|
|
fn truncate_error(e: &str) -> String {
|
|
if e.len() <= MAX_ERROR_LEN {
|
|
return e.to_owned();
|
|
}
|
|
let cut = (0..=MAX_ERROR_LEN)
|
|
.rev()
|
|
.find(|i| e.is_char_boundary(*i))
|
|
.unwrap_or(0);
|
|
let mut msg = e[..cut].to_owned();
|
|
msg.push('…');
|
|
msg
|
|
}
|