diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 12522f18..8181995f 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -363,17 +363,6 @@ impl JobQueue { .filter_map(|c| dag_view(&inner, c)) .collect() } - - /// Number of live (non-terminal) DAGs — tests + diagnostics. - #[cfg(test)] - #[must_use] - pub fn live_count(&self) -> usize { - let inner = self.lock(); - containers(&inner) - .into_iter() - .filter(|&c| inner.graph().is_settled(c) == Some(false)) - .count() - } } /// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals @@ -535,19 +524,34 @@ fn containers(sched: &Sched) -> Vec { /// this filter is what bounds what the dashboard sees. fn visible_dags(sched: &Sched) -> Vec { let mut live: Vec = Vec::new(); - let mut terminal: Vec<(NodeId, i64)> = Vec::new(); + let mut terminal: Vec<(NodeId, i64, u64)> = Vec::new(); for c in containers(sched) { if sched.graph().is_settled(c) == Some(true) { - terminal.push((c, dag_finished_at(sched, c))); + terminal.push((c, dag_finished_at(sched, c), c.get())); } else { live.push(c); } } + retain_history(live, terminal, MAX_HISTORY_DAGS) +} + +/// [`visible_dags`]'s policy, split from the graph it reads: keep every live +/// DAG, plus the newest `cap` terminal ones. +/// +/// `terminal` rows are `(handle, finished_at, tiebreak)`. The tiebreak orders +/// DAGs that settled inside the same wall-clock second — which is *most* of +/// them under a burst, and all of them in a test, so it is load-bearing rather +/// than a formality. +/// +/// Generic over the handle purely so this is reachable without a graph: a +/// `NodeId` cannot be fabricated, so a test that had to pass real ones could +/// only get them by submitting and running DAGs. +fn retain_history(live: Vec, mut terminal: Vec<(T, i64, u64)>, cap: usize) -> Vec { // 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); + terminal.sort_by(|a, b| b.1.cmp(&a.1).then(b.2.cmp(&a.2))); + terminal.truncate(cap); let mut kept = live; - kept.extend(terminal.into_iter().map(|(c, _)| c)); + kept.extend(terminal.into_iter().map(|(handle, _, _)| handle)); kept } diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 9d8874dd..198b06bf 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -405,29 +405,6 @@ pub fn approval_deploy( } } -/// A single `Reconcile` node that converges observed power state to the -/// persisted intent — `wanted` is untouched (no `SetWanted`), unlike the -/// operator `start`/`stop` templates. Test-only helper now (used to build -/// single-node lifecycle DAGs that exercise per-agent lease serialization -/// in the queue tests); production paths no longer emit a bare reconcile. -#[cfg(test)] -pub fn reconcile_only( - agent: &str, - source: Source, - reason: String, -) -> DagSpec> { - let agent = agent.to_owned(); - DagSpec { - source, - reason, - declare: Box::new(move |b: &Job| { - // Name the lease before the agent string is moved into the kind. - let lease = Resource::Agent(agent.clone()); - let _reconcile = b.node(NodeKind::Reconcile { agent }).needs(lease); - }), - } -} - /// First-deploy spawn (approval-driven): `Provision` (proposed/applied /// repos, state subvolume, meta registration) then `Create` /// (`nixos-container create`), drop-in write, then `Reconcile` starts diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 791a97f1..ae94f24c 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1451,36 +1451,59 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() { /// The dashboard renders one recent-builds list, so one number bounds it — /// and with no bucketing there's nothing for a burst of same-shaped DAGs to /// evict early, which is what the grace window used to paper over. +/// +/// `retain_history` is a sort-and-truncate over `(handle, finished_at, +/// tiebreak)`. This used to submit `MAX_HISTORY_DAGS + 8` DAGs, claim and fail +/// each one's node, then read the ids back out of a snapshot — the scheduler, +/// the roll-up and the wire projection all in the path of a policy that reads +/// none of them. +/// +/// Calling it directly also reaches the half the round-trip never could: every +/// DAG in that loop settled inside the same wall-clock second, so `finished_at` +/// tied on all of them and **only** the tiebreak was ever exercised. Eviction +/// by time — the actual policy — went untested. #[test] -fn history_evicts_oldest_terminals_past_flat_cap() { - const OVERFLOW: usize = 8; - let q = JobQueue::new(1); - let mut ids = Vec::new(); - for i in 0..(MAX_HISTORY_DAGS + OVERFLOW) { - let id = submit( - &q, - templates::reconcile_only(&format!("agent-{i}"), Source::Manual, "start".to_owned()), - ); - let c = claim_one(&q); - // Fail the single work node so the DAG *lingers*: a fully-`Done` DAG - // drops off the wire entirely, but a `Failed` one is retained (+ - // history-capped) so the operator can still triage it. Completing the - // node rolls the container up terminal. - q.complete_node(c.node_id, Err("boom".to_owned())); - ids.push(id); - } - let kept: std::collections::HashSet = q.snapshot().iter().map(|d| d.id).collect(); - assert_eq!(kept.len(), MAX_HISTORY_DAGS, "flat history cap"); - // Newest-first: the oldest `OVERFLOW` fall off, everything after survives. - // These DAGs settle within the same wall-clock second, so this also pins - // the `NodeId`-descending tiebreak that orders them when `finished_at` ties. - for old in &ids[..OVERFLOW] { - assert!(!kept.contains(old), "oldest terminal {old} evicted"); - } - for recent in &ids[OVERFLOW..] { - assert!(kept.contains(recent), "recent terminal {recent} retained"); - } - assert_eq!(q.live_count(), 0); +fn history_retains_live_dags_and_the_newest_terminals() { + const CAP: usize = 3; + // Live DAGs are kept whole, regardless of the cap. + assert_eq!( + retain_history(vec!["live-a", "live-b"], vec![], CAP), + vec!["live-a", "live-b"], + ); + // Terminals: newest `finished_at` first, oldest evicted past the cap. + assert_eq!( + retain_history( + vec![], + vec![ + ("oldest", 100, 1), + ("newest", 400, 2), + ("middle", 200, 3), + ("later", 300, 4), + ], + CAP, + ), + vec!["newest", "later", "middle"], + "the oldest terminal falls off" + ); + // Same second → the tiebreak decides, descending, so the DAG inserted + // last wins. This is the *common* case: a burst settles together. + assert_eq!( + retain_history( + vec![], + vec![("a", 100, 1), ("b", 100, 2), ("c", 100, 3), ("d", 100, 4)], + CAP, + ), + vec!["d", "c", "b"], + ); + // A live DAG never competes with history for the cap. + assert_eq!( + retain_history( + vec!["live"], + vec![("a", 100, 1), ("b", 200, 2), ("c", 300, 3), ("d", 400, 4)], + CAP, + ), + vec!["live", "d", "c", "b"], + ); } #[test]