refactor(#2756): declare the terminal hook instead of inferring it
`Template` was a DAG-level enum that three different things read back out: `terminal_hook()` mapped it to a side effect, the retention pass bucketed history by it, and a tracing field printed it. None of those needed a *label* — they needed the two facts the label happened to encode. So the enum was a lossy stand-in for intent, and every new DAG shape had to pick the variant whose inferred behaviour matched, whether or not the name fit (`reparent` rode `MetaUpdate` for exactly this reason, with a 10-line comment apologising for it). Replace the inference with a declaration: `DagSpec.hook: Option<HookKind>`. Only the builder assembling a DAG knows why it did so, so only the builder can say what should happen when it settles. `run_terminal_hook` becomes a field read, and `reparent`'s apology becomes `hook: None`. Hook assignment is byte-identical to the old precedence rule (`approval_id.is_some()` wins, then `Rebuild | PermChange`), checked site by site; `meta_update` is the only builder with a variable approval id and so the only remaining conditional. Retention loses the per-template bucket with the enum that keyed it. The dashboard renders one recent-builds list, so one flat newest-first cap (`MAX_HISTORY_DAGS`) bounds it. `HISTORY_GRACE_SECS` goes too — it existed to stop a burst of same-template DAGs evicting each other inside one poll interval, which is not a failure mode a flat cap has. That takes `snapshot_capped()` and the `snapshot_no_grace()` test hook with it. The queue is runtime-only (empty graph on boot), so the serde changes carry no migration risk.
This commit is contained in:
atlas
mara
parent
a8728ac532
commit
ca7146e4f0
8 changed files with 113 additions and 241 deletions
|
|
@ -11,15 +11,15 @@
|
|||
//! 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
|
||||
//! 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 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.
|
||||
//! container rolls up terminal, off the [`HookKind`] the *builder* stated on
|
||||
//! the spec: approval-resolve or `Rebuilt`-emit. 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
|
||||
|
|
@ -48,18 +48,14 @@ use tokio::sync::Notify;
|
|||
|
||||
use crate::coordinator::TransientKind;
|
||||
pub use model::{
|
||||
DagSpec, DagView, DepWhen, NodeKind, NodeSpec, PermPayload, Source, State, Template,
|
||||
DagSpec, DagView, DepWhen, HookKind, NodeKind, NodeSpec, PermPayload, Source, State,
|
||||
};
|
||||
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;
|
||||
/// 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;
|
||||
|
|
@ -74,7 +70,6 @@ pub struct Claim {
|
|||
/// 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
|
||||
|
|
@ -88,7 +83,8 @@ pub struct Claim {
|
|||
/// revert). Computed on demand from live graph state, not drained.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct TerminalDag {
|
||||
pub template: Template,
|
||||
/// The side effect to fire, carried from the DAG container's payload.
|
||||
pub hook: Option<HookKind>,
|
||||
/// Distinct agents this DAG's nodes targeted (one for a single-agent DAG).
|
||||
pub agents: Vec<String>,
|
||||
pub approval_id: Option<i64>,
|
||||
|
|
@ -110,7 +106,7 @@ struct NodeRuntime {
|
|||
/// 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,
|
||||
hook: Option<HookKind>,
|
||||
source: Source,
|
||||
reason: String,
|
||||
transient: Option<TransientKind>,
|
||||
|
|
@ -161,37 +157,6 @@ fn to_crate_when(when: DepWhen) -> JobDepWhen {
|
|||
}
|
||||
}
|
||||
|
||||
/// 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,
|
||||
}
|
||||
|
||||
/// The terminal hook a DAG needs, from its template + approval id — or `None`
|
||||
/// for a DAG with no terminal side effect (power-op, meta-update, boot, bare
|
||||
/// reconcile).
|
||||
///
|
||||
/// A cancelled DAG deliberately gets **no** compensating hook. [`JobQueue::cancel`]
|
||||
/// refuses unless every work node is still `Pending`, and a cancel *cascade*
|
||||
/// rolls up `Failed` (see `dag_rollup`), never `Cancelled` — so on a
|
||||
/// `Cancelled` DAG no node ever executed and there is nothing to undo. A power
|
||||
/// op's `SetWanted` head provably never ran, so its intent is still whatever
|
||||
/// the operator last set it to.
|
||||
#[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),
|
||||
_ => 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 {
|
||||
|
|
@ -289,7 +254,7 @@ impl JobQueue {
|
|||
.sched
|
||||
.append(
|
||||
NodeKind::Dag {
|
||||
template: spec.template,
|
||||
hook: spec.hook,
|
||||
source: spec.source,
|
||||
reason: spec.reason,
|
||||
transient: spec.transient,
|
||||
|
|
@ -380,7 +345,6 @@ impl JobQueue {
|
|||
node_id: id,
|
||||
kind,
|
||||
agent,
|
||||
template: meta.template,
|
||||
approval_id: meta.approval_id,
|
||||
inputs: meta.inputs,
|
||||
transient: meta.transient,
|
||||
|
|
@ -535,14 +499,8 @@ impl JobQueue {
|
|||
/// 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);
|
||||
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()
|
||||
}
|
||||
|
|
@ -558,14 +516,6 @@ impl JobQueue {
|
|||
.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 {
|
||||
|
|
@ -616,7 +566,7 @@ impl QueueInner {
|
|||
/// not a stored side-table.
|
||||
fn dag_meta(&self, container: NodeId) -> Option<DagMeta> {
|
||||
let NodeKind::Dag {
|
||||
template,
|
||||
hook,
|
||||
source,
|
||||
reason,
|
||||
transient,
|
||||
|
|
@ -628,7 +578,7 @@ impl QueueInner {
|
|||
return None;
|
||||
};
|
||||
Some(DagMeta {
|
||||
template: *template,
|
||||
hook: *hook,
|
||||
source: *source,
|
||||
reason: reason.clone(),
|
||||
transient: *transient,
|
||||
|
|
@ -708,7 +658,7 @@ impl QueueInner {
|
|||
fn terminal_dag(&self, container: NodeId) -> Option<TerminalDag> {
|
||||
let meta = self.dag_meta(container)?;
|
||||
Some(TerminalDag {
|
||||
template: meta.template,
|
||||
hook: meta.hook,
|
||||
agents: self.dag_agents(container),
|
||||
approval_id: meta.approval_id,
|
||||
state: self.dag_rollup(container),
|
||||
|
|
@ -829,33 +779,24 @@ impl QueueInner {
|
|||
}
|
||||
|
||||
/// 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> {
|
||||
/// 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, Template, i64)> = Vec::new();
|
||||
let mut terminal: Vec<(NodeId, 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)));
|
||||
}
|
||||
terminal.push((c, 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();
|
||||
// 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;
|
||||
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.extend(terminal.into_iter().map(|(c, _)| c));
|
||||
kept
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue