03eb64cb5c
Node gains started_at/finished_at (chrono DateTime<Utc>, serialized RFC 3339 on the wire per hive_sh4re::wire_time) plus error (String). Graph::set_state self-stamps started_at on the first Running transition and finished_at on the first terminal one, via an internal now_utc() clock (keeps settle/complete signatures stable). Outcome::Failed(String) carries the failure reason, set on the terminal transition. hive-c0re complete_node builds Outcome::Failed(msg); its node_rt side-table stays i64 for now (double-write) until #2637 reads the Node. Toward #2637: the jobq graph becomes the source of truth for per-node lifecycle so the queue can be sent to the client as-is.
882 lines
34 KiB
Rust
882 lines
34 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 template's 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 runs **inline** ([`exec::run_terminal_hook`]) when the
|
|
//! container rolls up terminal — dispatched off its template
|
|
//! ([`terminal_hook`]): approval-resolve, `Rebuilt`-emit, or power-intent
|
|
//! revert. No terminal-hook node, no drained event stream.
|
|
//!
|
|
//! 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 hive_jobq::resources::ResourceTable;
|
|
use hive_jobq::scheduler::{Outcome, Scheduler};
|
|
use hive_jobq::{Dep, DepWhen as JobDepWhen, Graph, NodeId, State as JobState};
|
|
use hive_sh4re::jobs::NodeView;
|
|
use hive_sh4re::wire_time::now_unix;
|
|
use tokio::sync::Notify;
|
|
|
|
use crate::coordinator::TransientKind;
|
|
pub use model::{
|
|
DagSpec, DagView, DepWhen, NodeKind, NodeSpec, PermPayload, Source, State, Template,
|
|
};
|
|
use resource::Resource;
|
|
|
|
/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) to retain per
|
|
/// template in the snapshot, matching the old per-kind history cap.
|
|
const MAX_HISTORY_PER_TEMPLATE: usize = 5;
|
|
|
|
/// Terminal DAGs younger than this are exempt from the per-template history
|
|
/// cap, so a burst of same-template DAGs that settle within one `QueueDag`
|
|
/// poll interval isn't evicted before the poller observes their terminal state.
|
|
const HISTORY_GRACE_SECS: i64 = 300;
|
|
|
|
/// Cap on stored node error strings.
|
|
const MAX_ERROR_LEN: usize = 2_000;
|
|
|
|
/// 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,
|
|
pub template: Template,
|
|
pub approval_id: Option<i64>,
|
|
pub inputs: Vec<String>,
|
|
/// Transient pill kind for the lease window (from the spec). Whether the
|
|
/// pill is currently shown is derived from live lease ownership
|
|
/// ([`JobQueue::held_transients`]), not a per-claim edge.
|
|
pub transient: Option<TransientKind>,
|
|
}
|
|
|
|
/// Summary of a DAG's terminal roll-up — the input to the terminal node's
|
|
/// executor (approval resolution, `Rebuilt` emission, cancelled-power-op intent
|
|
/// revert). Computed on demand from live graph state, not drained.
|
|
#[derive(Debug, Clone)]
|
|
pub struct TerminalDag {
|
|
pub template: Template,
|
|
/// Distinct agents this DAG's nodes targeted (one for a single-agent DAG).
|
|
pub agents: Vec<String>,
|
|
pub approval_id: Option<i64>,
|
|
pub state: State,
|
|
/// First failed node's error when `state == Failed`.
|
|
pub error: Option<String>,
|
|
}
|
|
|
|
/// Per-node runtime metadata the crate graph doesn't carry (kind + agent live
|
|
/// in the node payload; state lives in the node).
|
|
#[derive(Debug, Default, Clone)]
|
|
struct NodeRuntime {
|
|
step: Option<String>,
|
|
build_log_id: Option<i64>,
|
|
started_at: Option<i64>,
|
|
finished_at: Option<i64>,
|
|
error: Option<String>,
|
|
}
|
|
|
|
/// 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 {
|
|
template: Template,
|
|
source: Source,
|
|
reason: String,
|
|
transient: Option<TransientKind>,
|
|
approval_id: Option<i64>,
|
|
inputs: Vec<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::subtree`] /
|
|
/// [`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.
|
|
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)
|
|
}
|
|
}
|
|
|
|
/// Map a spec dependency edge kind onto the crate's.
|
|
fn to_crate_when(when: DepWhen) -> JobDepWhen {
|
|
match when {
|
|
DepWhen::AfterOk => JobDepWhen::AfterOk,
|
|
DepWhen::AfterAny => JobDepWhen::AfterAny,
|
|
}
|
|
}
|
|
|
|
/// The inline terminal-hook a settled DAG fires — dispatched off its container's
|
|
/// template + approval id when the container rolls up terminal (no hook node).
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum HookKind {
|
|
/// Approval-driven DAG (spawn / opaque deploy): resolve the approval row.
|
|
ResolveApproval,
|
|
/// Rebuild / perm-change: emit one `Rebuilt` manager event per agent.
|
|
EmitRebuilt,
|
|
/// Power-op: on a *cancelled* DAG, revert each agent's `wanted` intent.
|
|
RevertIntent,
|
|
}
|
|
|
|
/// The terminal hook a DAG needs, from its template + approval id — or `None`
|
|
/// for a DAG with no terminal side effect (meta-update, boot, bare reconcile).
|
|
#[must_use]
|
|
pub fn terminal_hook(template: Template, approval_id: Option<i64>) -> Option<HookKind> {
|
|
if approval_id.is_some() {
|
|
return Some(HookKind::ResolveApproval);
|
|
}
|
|
match template {
|
|
Template::Rebuild | Template::PermChange => Some(HookKind::EmitRebuilt),
|
|
Template::Start
|
|
| Template::Stop
|
|
| Template::GracefulStop
|
|
| Template::Restart
|
|
| Template::GracefulRestart => Some(HookKind::RevertIntent),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
/// Map a crate node state onto the wire state (`Pending` ↔ `Queued`;
|
|
/// `Finishing` — own logic done, sub-nodes still running — reads as `Running`).
|
|
fn to_wire_state(state: JobState) -> State {
|
|
match state {
|
|
JobState::Pending => State::Queued,
|
|
JobState::Running | JobState::Finishing => State::Running,
|
|
JobState::Done => State::Done,
|
|
JobState::Failed => State::Failed,
|
|
JobState::Cancelled => State::Cancelled,
|
|
}
|
|
}
|
|
|
|
/// 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: to_crate_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 and it
|
|
/// reaching terminal fires the DAG's inline hook.
|
|
///
|
|
/// # 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 {
|
|
template: spec.template,
|
|
source: spec.source,
|
|
reason: spec.reason,
|
|
transient: spec.transient,
|
|
approval_id: spec.approval_id,
|
|
inputs: spec.inputs,
|
|
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 now = now_unix();
|
|
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.dag_of(id) else {
|
|
continue;
|
|
};
|
|
let Some(meta) = inner.dag_meta(container) else {
|
|
continue;
|
|
};
|
|
claims.push(Claim {
|
|
dag_id: container.get(),
|
|
node_id: id,
|
|
kind,
|
|
agent,
|
|
template: meta.template,
|
|
approval_id: meta.approval_id,
|
|
inputs: meta.inputs,
|
|
transient: meta.transient,
|
|
});
|
|
if let Some(rt) = inner.node_rt.get_mut(&id) {
|
|
rt.started_at = Some(now);
|
|
}
|
|
}
|
|
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. Returns the DAG's
|
|
/// terminal summary **iff** this completion rolled its container terminal —
|
|
/// the scheduler runs the DAG's inline hook off it.
|
|
pub fn complete_node(
|
|
&self,
|
|
_dag_id: u64,
|
|
node_id: NodeId,
|
|
result: Result<(), String>,
|
|
) -> Option<TerminalDag> {
|
|
let mut inner = self.lock();
|
|
let now = now_unix();
|
|
let (error, outcome) = match result {
|
|
Ok(()) => (None, Outcome::Done),
|
|
Err(e) => {
|
|
// The reason rides the crate `Outcome::Failed` (stamped onto the
|
|
// graph `Node`); the `node_rt` copy stays for now until the wire
|
|
// reads it off the node directly.
|
|
let msg = truncate_error(&e);
|
|
(Some(msg.clone()), Outcome::Failed(msg))
|
|
}
|
|
};
|
|
if let Some(rt) = inner.node_rt.get_mut(&node_id) {
|
|
rt.finished_at = Some(now);
|
|
rt.step = None;
|
|
if let Some(e) = error {
|
|
rt.error = Some(e);
|
|
}
|
|
}
|
|
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,
|
|
// hand its summary back so the scheduler fires the inline hook once.
|
|
let terminal = container
|
|
.filter(|&c| c != node_id && inner.dag_is_terminal(c))
|
|
.and_then(|c| inner.terminal_dag(c));
|
|
drop(inner);
|
|
self.notify.notify_one();
|
|
terminal
|
|
}
|
|
|
|
/// Cancel a DAG that hasn't started yet: every work node is still `Pending`,
|
|
/// so each is cancelled. `None` once any work node is running or terminal —
|
|
/// an in-flight nix build isn't interruptible. Otherwise the container is
|
|
/// rolled up so the DAG settles (wire state `Cancelled`) and its terminal
|
|
/// summary is returned — the caller fires the inline hook (power-intent
|
|
/// revert / approval resolution) off it.
|
|
pub fn cancel(&self, dag_id: u64) -> Option<TerminalDag> {
|
|
let mut inner = self.lock();
|
|
let container = inner.container(dag_id)?;
|
|
let work = inner.subtree(container);
|
|
let all_pending = work.iter().all(|&id| {
|
|
inner
|
|
.sched
|
|
.graph()
|
|
.node(id)
|
|
.is_some_and(|n| n.state == JobState::Pending)
|
|
});
|
|
if !all_pending {
|
|
return None;
|
|
}
|
|
for id in work {
|
|
inner.sched.cancel_node(id);
|
|
}
|
|
// The container was settled to `Finishing` at submit; completing it again
|
|
// now re-runs the roll-up with its children all `Cancelled`, driving it to
|
|
// a terminal state synchronously within this lock — so the caller reads
|
|
// the terminal summary immediately instead of waiting for the scheduler
|
|
// loop to observe the cancellation. `dag_rollup` reports `Cancelled` to
|
|
// the wire (a container whose children all cancelled).
|
|
inner.sched.complete(container, Outcome::Done);
|
|
let terminal = inner.terminal_dag(container);
|
|
drop(inner);
|
|
self.notify.notify_one();
|
|
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();
|
|
if inner.dag_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
|
|
}
|
|
|
|
/// Link a `build_logs` row to the DAG's currently-running node — DAG-id-only
|
|
/// compatibility surface (approval pipeline callbacks).
|
|
pub fn set_build_log_id_running(&self, dag_id: u64, log_id: i64) -> bool {
|
|
let mut inner = self.lock();
|
|
let Some(node_id) = inner.running_node_of(dag_id) else {
|
|
return false;
|
|
};
|
|
inner.node_rt.entry(node_id).or_default().build_log_id = Some(log_id);
|
|
true
|
|
}
|
|
|
|
/// A DAG's terminal roll-up summary, computed on demand from its container.
|
|
/// `None` if the DAG id is unknown. Test-only — production reads the summary
|
|
/// `complete_node` returns when the container rolls up terminal.
|
|
#[cfg(test)]
|
|
#[must_use]
|
|
pub(crate) fn terminal_summary(&self, dag_id: u64) -> Option<TerminalDag> {
|
|
let inner = self.lock();
|
|
let container = inner.container(dag_id)?;
|
|
inner.terminal_dag(container)
|
|
}
|
|
|
|
/// The `(dag_id, agent, kind)` triples for every per-agent lease currently
|
|
/// held by a DAG that carries a transient pill — the live transient-pill
|
|
/// set, a pull query over crate resource ownership (replaces the old
|
|
/// lease-release event stream). A DAG with no transient kind is omitted.
|
|
#[must_use]
|
|
pub fn held_transients(&self) -> Vec<(u64, String, TransientKind)> {
|
|
let inner = self.lock();
|
|
inner
|
|
.sched
|
|
.resource_state()
|
|
.into_iter()
|
|
.filter_map(|(res, holder)| {
|
|
let Resource::Agent(agent) = res else {
|
|
return None;
|
|
};
|
|
let container = inner.dag_of(holder)?;
|
|
let kind = inner.dag_meta(container)?.transient?;
|
|
Some((container.get(), agent, kind))
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`.
|
|
#[must_use]
|
|
pub fn snapshot(&self) -> Vec<DagView> {
|
|
self.snapshot_capped(now_unix() - HISTORY_GRACE_SECS)
|
|
}
|
|
|
|
/// Snapshot the visible DAG set (live + newest-per-template terminal, terminal
|
|
/// ones after `grace_cutoff` always kept), sorted by container id.
|
|
fn snapshot_capped(&self, grace_cutoff: i64) -> Vec<DagView> {
|
|
let inner = self.lock();
|
|
let mut ids = inner.visible_dags(grace_cutoff);
|
|
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.dag_is_terminal(c))
|
|
.count()
|
|
}
|
|
|
|
/// Test hook: snapshot with the history grace window disabled, so the
|
|
/// per-template cap applies to just-finished terminal DAGs too.
|
|
#[cfg(test)]
|
|
#[must_use]
|
|
pub(crate) fn snapshot_no_grace(&self) -> Vec<DagView> {
|
|
self.snapshot_capped(i64::MAX)
|
|
}
|
|
}
|
|
|
|
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 == JobState::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 DAG container a node belongs to — walk its parent chain to the root
|
|
/// (`parent == None`), which is the container. Returns `id` itself for a
|
|
/// container node.
|
|
fn dag_of(&self, id: NodeId) -> Option<NodeId> {
|
|
let mut cur = id;
|
|
loop {
|
|
match self.sched.graph().node(cur)?.parent {
|
|
Some(p) => cur = p,
|
|
None => return Some(cur),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The DAG's work nodes — its `container`'s subtree, excluding the container.
|
|
fn subtree(&self, container: NodeId) -> Vec<NodeId> {
|
|
self.sched
|
|
.graph()
|
|
.nodes()
|
|
.filter(|n| n.id != container && self.dag_of(n.id) == Some(container))
|
|
.map(|n| n.id)
|
|
.collect()
|
|
}
|
|
|
|
/// 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 {
|
|
template,
|
|
source,
|
|
reason,
|
|
transient,
|
|
approval_id,
|
|
inputs,
|
|
created_at,
|
|
} = &self.sched.graph().node(container)?.payload
|
|
else {
|
|
return None;
|
|
};
|
|
Some(DagMeta {
|
|
template: *template,
|
|
source: *source,
|
|
reason: reason.clone(),
|
|
transient: *transient,
|
|
approval_id: *approval_id,
|
|
inputs: inputs.clone(),
|
|
created_at: *created_at,
|
|
})
|
|
}
|
|
|
|
/// 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
|
|
/// child shows `Failed` immediately, before the container finishes rolling
|
|
/// up — matching the pre-container behaviour.)
|
|
fn dag_rollup(&self, container: NodeId) -> State {
|
|
let mut any_running = false;
|
|
let mut any_queued = false;
|
|
let mut any_cancelled = false;
|
|
for id in self.subtree(container) {
|
|
match self.sched.graph().node(id).map(|n| n.state) {
|
|
Some(JobState::Failed) => return State::Failed,
|
|
Some(JobState::Running | JobState::Finishing) => any_running = true,
|
|
Some(JobState::Pending) => any_queued = true,
|
|
Some(JobState::Cancelled) => any_cancelled = true,
|
|
Some(JobState::Done) | None => {}
|
|
}
|
|
}
|
|
if any_running {
|
|
State::Running
|
|
} else if any_queued {
|
|
State::Queued
|
|
} else if any_cancelled {
|
|
State::Cancelled
|
|
} else {
|
|
State::Done
|
|
}
|
|
}
|
|
|
|
/// True when the DAG has settled — its container has rolled up terminal
|
|
/// (equivalent to every work node being terminal).
|
|
fn dag_is_terminal(&self, container: NodeId) -> bool {
|
|
self.sched
|
|
.graph()
|
|
.node(container)
|
|
.is_some_and(|n| n.state.is_terminal())
|
|
}
|
|
|
|
/// Distinct agents a DAG's work nodes target, in first-seen order.
|
|
fn dag_agents(&self, container: NodeId) -> Vec<String> {
|
|
let mut seen: Vec<String> = Vec::new();
|
|
for id in self.subtree(container) {
|
|
if let Some(n) = self.sched.graph().node(id) {
|
|
let agent = n.payload.agent();
|
|
if !agent.is_empty() && !seen.iter().any(|s| s == agent) {
|
|
seen.push(agent.to_owned());
|
|
}
|
|
}
|
|
}
|
|
seen
|
|
}
|
|
|
|
/// First failed work node's stored error, for the roll-up `error` field.
|
|
fn dag_first_error(&self, container: NodeId) -> Option<String> {
|
|
for id in self.subtree(container) {
|
|
if self
|
|
.sched
|
|
.graph()
|
|
.node(id)
|
|
.is_some_and(|n| n.state == JobState::Failed)
|
|
&& let Some(e) = self.node_rt.get(&id).and_then(|r| r.error.clone())
|
|
{
|
|
return Some(e);
|
|
}
|
|
}
|
|
None
|
|
}
|
|
|
|
/// A DAG's terminal roll-up summary — the input to its inline hook.
|
|
fn terminal_dag(&self, container: NodeId) -> Option<TerminalDag> {
|
|
let meta = self.dag_meta(container)?;
|
|
Some(TerminalDag {
|
|
template: meta.template,
|
|
agents: self.dag_agents(container),
|
|
approval_id: meta.approval_id,
|
|
state: self.dag_rollup(container),
|
|
error: self.dag_first_error(container),
|
|
})
|
|
}
|
|
|
|
/// Rebuild the wire [`DagView`] for a DAG from its container metadata + work
|
|
/// nodes + per-node runtime.
|
|
fn dag_view(&self, container: NodeId) -> Option<DagView> {
|
|
let meta = self.dag_meta(container)?;
|
|
let node_ids = self.subtree(container);
|
|
let mut nodes = Vec::with_capacity(node_ids.len());
|
|
let mut started: Vec<i64> = Vec::new();
|
|
let mut finished: Vec<i64> = Vec::new();
|
|
for &id in &node_ids {
|
|
let Some(node) = self.sched.graph().node(id) else {
|
|
continue;
|
|
};
|
|
let rt = self.node_rt.get(&id);
|
|
let deps: Vec<u64> = node
|
|
.deps
|
|
.iter()
|
|
.filter_map(|d| match d {
|
|
Dep::Node { id, .. } => Some(id.get()),
|
|
Dep::Resource { .. } => None,
|
|
})
|
|
.collect();
|
|
if let Some(s) = rt.and_then(|r| r.started_at) {
|
|
started.push(s);
|
|
}
|
|
if let Some(fin) = rt.and_then(|r| r.finished_at) {
|
|
finished.push(fin);
|
|
}
|
|
nodes.push(NodeView {
|
|
id: id.get(),
|
|
agent: node.payload.agent().to_owned(),
|
|
kind: node.payload.as_str().to_owned(),
|
|
deps,
|
|
state: to_wire_state(node.state),
|
|
step: rt.and_then(|r| r.step.clone()),
|
|
build_log_id: rt.and_then(|r| r.build_log_id),
|
|
started_at: rt.and_then(|r| r.started_at),
|
|
finished_at: rt.and_then(|r| r.finished_at),
|
|
error: rt.and_then(|r| r.error.clone()),
|
|
});
|
|
}
|
|
let is_terminal = self.dag_is_terminal(container);
|
|
Some(DagView {
|
|
id: container.get(),
|
|
kind: meta.template,
|
|
state: self.dag_rollup(container),
|
|
source: meta.source,
|
|
reason: meta.reason.clone(),
|
|
enqueued_at: meta.created_at,
|
|
started_at: started.into_iter().min(),
|
|
finished_at: if is_terminal {
|
|
finished.into_iter().max()
|
|
} else {
|
|
None
|
|
},
|
|
inputs: meta.inputs.clone(),
|
|
approval_id: meta.approval_id,
|
|
nodes,
|
|
})
|
|
}
|
|
|
|
/// When a DAG's work node finishes on `finished_at` — the max over its
|
|
/// subtree, for the history cap ordering.
|
|
fn dag_finished_at(&self, container: NodeId) -> i64 {
|
|
self.subtree(container)
|
|
.iter()
|
|
.filter_map(|id| self.node_rt.get(id).and_then(|r| r.finished_at))
|
|
.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_PER_TEMPLATE`] terminal DAGs per template
|
|
/// (terminal DAGs finished after `grace_cutoff` are always kept). 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, grace_cutoff: i64) -> Vec<NodeId> {
|
|
let mut live: Vec<NodeId> = Vec::new();
|
|
let mut terminal: Vec<(NodeId, Template, i64)> = Vec::new();
|
|
for c in self.containers() {
|
|
if self.dag_is_terminal(c) {
|
|
if let Some(meta) = self.dag_meta(c) {
|
|
terminal.push((c, meta.template, self.dag_finished_at(c)));
|
|
}
|
|
} else {
|
|
live.push(c);
|
|
}
|
|
}
|
|
// Newest first so the per-template cap keeps the most recent.
|
|
terminal.sort_by(|a, b| b.2.cmp(&a.2).then(b.0.get().cmp(&a.0.get())));
|
|
let mut counts: HashMap<Template, usize> = HashMap::new();
|
|
let mut kept = live;
|
|
for (c, template, finished) in terminal {
|
|
let n = counts.entry(template).or_insert(0);
|
|
*n += 1;
|
|
if *n <= MAX_HISTORY_PER_TEMPLATE || finished > grace_cutoff {
|
|
kept.push(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
|
|
}
|