diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index ad1a4ff2..4e4f0021 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -361,7 +361,6 @@ impl JobQueue { 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 { @@ -511,6 +510,17 @@ impl JobQueue { true } + /// The `build_logs` row id linked to `node_id`, if any — the lookup behind + /// the `GET /api/build-log/` query endpoint (the client fetches a + /// node's captured build output on demand rather than receiving it inline). + #[must_use] + pub fn build_log_id_of(&self, node_id: NodeId) -> Option { + self.lock() + .node_rt + .get(&node_id) + .and_then(|r| r.build_log_id) + } + /// 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. diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index b385e420..21c584e1 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -45,11 +45,14 @@ fn claim_one(q: &JobQueue) -> Claim { } fn state_of(q: &JobQueue, dag_id: u64) -> State { + // A fully-`Done` DAG drops out of the snapshot (its nodes are all + // excluded) — absence is the completion signal, so map it to `Done`. + // Otherwise derive the roll-up from the node set, exactly as every wire + // consumer does. q.snapshot() .iter() .find(|d| d.id == dag_id) - .expect("dag present") - .state + .map_or(State::Done, DagView::rollup_state) } // ---- submit (dedup removed — every submit is a fresh DAG) ---- @@ -664,7 +667,7 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() { q.complete_node(id, reconcile.node_id, Ok(())); let snap = q.snapshot(); let dag = snap.iter().find(|d| d.id == id).expect("dag"); - assert_eq!(dag.state, State::Failed, "roll-up failed"); + assert_eq!(dag.rollup_state(), State::Failed, "roll-up failed"); let by_kind = |k: &str| { dag.nodes .iter() @@ -872,10 +875,10 @@ fn set_step_only_on_running_and_signals_change() { ); assert!(q.set_step(id, c.node_id, "next phase")); assert!(q.set_step_running(id, "via running lookup")); + // `step` is host-side only now (off the wire); completion clears it + // internally, but there's no wire field to observe — the return-value + // contract above (running-gating + change signalling) is the behaviour. q.complete_node(id, c.node_id, Ok(())); - let snap = q.snapshot(); - let node = &snap.iter().find(|d| d.id == id).expect("dag").nodes[0]; - assert_eq!(node.step, None, "step cleared on completion"); } #[test] @@ -890,9 +893,13 @@ fn set_build_log_id_links_running_node() { assert!(q.set_build_log_id(id, c.node_id, 42)); assert!(q.set_build_log_id_running(id, 43)); q.complete_node(id, c.node_id, Ok(())); - let snap = q.snapshot(); - let node = &snap.iter().find(|d| d.id == id).expect("dag").nodes[0]; - assert_eq!(node.build_log_id, Some(43), "log id survives completion"); + // The log id is fetched by node id (the `GET /api/build-log/` lookup), + // not carried on the wire — it survives completion in the node runtime. + assert_eq!( + q.build_log_id_of(c.node_id), + Some(43), + "log id survives completion" + ); } #[test] diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 4bfb3bf4..d117f977 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -795,9 +795,14 @@ async fn await_dags(coord: &Arc, ids: &[u64], timeout: std::time::D let deadline = std::time::Instant::now() + timeout; loop { let snap = coord.job_queue.snapshot(); - let pending = ids - .iter() - .any(|id| snap.iter().any(|d| d.id == *id && !d.state.is_terminal())); + // A DAG has settled when it's either gone from the snapshot (fully + // `Done` DAGs drop out) or still present but with every node terminal + // (a `Failed`/`Cancelled` DAG lingers). It's pending only while it has + // a non-terminal node. + let pending = ids.iter().any(|id| { + snap.iter() + .any(|d| d.id == *id && d.nodes.iter().any(|n| !n.state.is_terminal())) + }); if !pending { return; } diff --git a/hive-sh4re/src/jobs.rs b/hive-sh4re/src/jobs.rs index 10b4c5e0..caf2fc04 100644 --- a/hive-sh4re/src/jobs.rs +++ b/hive-sh4re/src/jobs.rs @@ -142,3 +142,37 @@ pub struct DagView { /// lingers until aged out by the history cap. pub nodes: Vec, } + +impl DagView { + /// Roll-up state derived from the node set — the shared derivation every + /// Rust consumer (hivectl, the wait loops, tests) uses so the dashboard's + /// JS render and the host agree: `Failed` if any node failed, else + /// `Running` if any running, else `Queued` if any queued, else + /// `Cancelled` if any cancelled, else `Done`. `Done` nodes are excluded + /// from the wire, so a DAG that is *entirely* done isn't sent at all — + /// its absence from the snapshot is what signals completion. + #[must_use] + pub fn rollup_state(&self) -> State { + let mut any_running = false; + let mut any_queued = false; + let mut any_cancelled = false; + for n in &self.nodes { + match n.state { + State::Failed => return State::Failed, + State::Running => any_running = true, + State::Queued => any_queued = true, + State::Cancelled => any_cancelled = true, + State::Done => {} + } + } + if any_running { + State::Running + } else if any_queued { + State::Queued + } else if any_cancelled { + State::Cancelled + } else { + State::Done + } + } +}