diff --git a/Cargo.lock b/Cargo.lock index a75fe8fc..eb4d9698 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1694,6 +1694,7 @@ dependencies = [ name = "hive-jobq" version = "0.1.0" dependencies = [ + "chrono", "serde", "serde_json", "thiserror 2.0.18", diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index e2068cb7..2e5a987a 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -408,7 +408,13 @@ impl JobQueue { let now = now_unix(); let (error, outcome) = match result { Ok(()) => (None, Outcome::Done), - Err(e) => (Some(truncate_error(&e)), Outcome::Failed), + 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); diff --git a/hive-jobq/Cargo.toml b/hive-jobq/Cargo.toml index 71e644a4..13059fa1 100644 --- a/hive-jobq/Cargo.toml +++ b/hive-jobq/Cargo.toml @@ -7,6 +7,7 @@ version.workspace = true workspace = true [dependencies] +chrono = { workspace = true } serde = { workspace = true } thiserror = { workspace = true } diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs index c4c05985..5963f042 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -29,6 +29,8 @@ pub mod resources; pub mod scheduler; +use chrono::{DateTime, Utc}; + /// Opaque, stable, monotonic node identifier. /// /// Assigned by the [`Graph`] on insert and persisted, so it is stable across @@ -141,6 +143,19 @@ impl State { } } +/// Wall-clock UTC now — the source for node lifecycle timestamps +/// ([`Node::started_at`] / [`Node::finished_at`]). The graph stamps its own +/// timestamps rather than threading a clock through every call, so a node's +/// timing is self-contained. Derived from `SystemTime` (the workspace `chrono` +/// carries no `clock` feature, matching `hive_sh4re::wire_time`), truncated to +/// whole seconds; a pre-epoch or out-of-range clock clamps to the epoch. +fn now_utc() -> DateTime { + let secs = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX)); + DateTime::::from_timestamp(secs, 0).unwrap_or_default() +} + /// A node in the graph, carrying a caller-defined payload `N`. /// /// The library schedules over `Node`s and resources without interpreting the @@ -165,6 +180,17 @@ pub struct Node { pub deps: Vec>, /// Lifecycle state. pub state: State, + /// UTC instant the node entered [`State::Running`] (`None` until it starts; + /// a cancelled node never ran, so it stays `None`). Stamped by the graph. + pub started_at: Option>, + /// UTC instant the node reached a terminal state (`Done` / `Failed` / + /// `Cancelled`). `None` while non-terminal. Stamped by the graph. + pub finished_at: Option>, + /// Failure reason for a `Failed` node, supplied by the runner via + /// [`scheduler::Outcome::Failed`]. `None` unless this node's own logic + /// failed (a node that rolled up `Failed` from a child, or was cancelled, + /// carries no error of its own). + pub error: Option, } /// An error from inserting into or loading a [`Graph`] with a dangling id. @@ -308,6 +334,9 @@ impl Graph { payload, deps, state: State::Pending, + started_at: None, + finished_at: None, + error: None, }); Ok(id) } @@ -353,16 +382,35 @@ impl Graph { /// Set a node's lifecycle state, returning `false` for an unknown id. The /// scheduler drives every state transition — nothing else mutates state, - /// which is what keeps the resource guards + terminality in sync. + /// which is what keeps the resource guards + terminality in sync. This is + /// also where the node's lifecycle timestamps are stamped: `started_at` on + /// the first transition to [`State::Running`], `finished_at` on the first + /// transition to a terminal state (`Done` / `Failed` / `Cancelled`). pub(crate) fn set_state(&mut self, id: NodeId, state: State) -> bool { if let Some(node) = self.nodes.iter_mut().find(|n| n.id == id) { node.state = state; + if state == State::Running { + if node.started_at.is_none() { + node.started_at = Some(now_utc()); + } + } else if state.is_terminal() && node.finished_at.is_none() { + node.finished_at = Some(now_utc()); + } true } else { false } } + /// Record a node's failure reason ([`Node::error`]). No-op for an unknown + /// id. Called by the scheduler on an [`scheduler::Outcome::Failed`] before + /// the terminal state transition. + pub(crate) fn set_error(&mut self, id: NodeId, error: String) { + if let Some(node) = self.nodes.iter_mut().find(|n| n.id == id) { + node.error = Some(error); + } + } + /// Check that every id the graph holds resolves: every [`Dep::Node`] id /// names a node present in the graph, and `next_id` is past the largest /// existing id. Deserialization runs this, so a loaded graph is internally @@ -568,6 +616,9 @@ mod tests { when: DepWhen::AfterOk, }], state: State::Pending, + started_at: None, + finished_at: None, + error: None, }], next_id: 1, }; @@ -585,6 +636,9 @@ mod tests { payload: "x", deps: vec![], state: State::Pending, + started_at: None, + finished_at: None, + error: None, }], next_id: 3, }; diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index 98dfd6b3..5cd80975 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -38,12 +38,14 @@ use crate::{Dep, DepWhen, Graph, GraphError, NodeId, State}; /// /// `Cancelled` is not an outcome a runner reports — it is scheduler-driven (an /// `AfterOk` dependency failed), so a runner only ever says `Done` or `Failed`. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] pub enum Outcome { /// The node's work succeeded. Done, - /// The node's work failed. - Failed, + /// The node's work failed, carrying the failure reason — recorded on + /// [`crate::Node::error`] for the failed node itself (a node that rolls up + /// `Failed` from a child, or is cancelled, carries no error of its own). + Failed(String), } /// Drives a [`Graph`] over an owned resource pool: claim runnable nodes, record @@ -211,7 +213,10 @@ impl Scheduler { /// to start newly-unblocked work. pub fn complete(&mut self, id: NodeId, outcome: Outcome) { match outcome { - Outcome::Failed => { + Outcome::Failed(error) => { + // Record the reason before the terminal transition so it's set + // by the time `set_state` stamps `finished_at`. + self.graph.set_error(id, error); self.graph.set_state(id, State::Failed); self.cascade_cancel(id); } @@ -454,6 +459,45 @@ mod tests { assert_eq!(avail(&s, "build-slot"), 1); } + #[test] + fn lifecycle_timestamps_and_error_are_stamped() { + let mut s = scheduler_with_slots(2); + let ok = s.append("ok", vec![], None).expect("insert"); + let bad = s.append("bad", vec![], None).expect("insert"); + let downstream = s.append("down", vec![after_ok(bad)], None).expect("insert"); + + // Before running: no timestamps. + assert!(s.graph().node(ok).unwrap().started_at.is_none()); + assert!(s.graph().node(ok).unwrap().finished_at.is_none()); + + let started = s.settle(); + assert!(started.contains(&ok) && started.contains(&bad)); + // Running → started_at stamped, finished_at still none. + assert!(s.graph().node(ok).unwrap().started_at.is_some()); + assert!(s.graph().node(ok).unwrap().finished_at.is_none()); + + // Done → finished_at stamped, no error. + s.complete(ok, Outcome::Done); + let n = s.graph().node(ok).unwrap(); + assert!(n.finished_at.is_some()); + assert_eq!(n.error, None); + + // Failed → the reason rides `Outcome::Failed`, finished_at stamped. + s.complete(bad, Outcome::Failed("boom".to_owned())); + let n = s.graph().node(bad).unwrap(); + assert_eq!(n.state, State::Failed); + assert_eq!(n.error.as_deref(), Some("boom")); + assert!(n.finished_at.is_some()); + + // The `AfterOk`-cancelled node: cascade-cancelled, so finished_at is set, + // but it never ran (no started_at) and carries no error of its own. + let n = s.graph().node(downstream).unwrap(); + assert_eq!(n.state, State::Cancelled); + assert!(n.started_at.is_none()); + assert!(n.finished_at.is_some()); + assert_eq!(n.error, None); + } + #[test] fn build_slot_cap_limits_concurrency_and_release_unblocks() { let mut s = scheduler_with_slots(2); @@ -514,7 +558,7 @@ mod tests { assert_eq!(s.settle(), vec![root]); s.complete(root, Outcome::Done); assert_eq!(s.settle(), vec![child]); - s.complete(child, Outcome::Failed); + s.complete(child, Outcome::Failed(String::new())); assert_eq!( s.graph().node(root).unwrap().state, State::Failed, @@ -689,7 +733,7 @@ mod tests { ) .expect("weak"); assert_eq!(s.settle(), vec![root]); - s.complete(root, Outcome::Failed); + s.complete(root, Outcome::Failed(String::new())); assert_eq!(s.graph().node(strong1).unwrap().state, State::Cancelled); assert_eq!(s.graph().node(strong2).unwrap().state, State::Cancelled); assert_eq!(s.settle(), vec![weak]); @@ -704,7 +748,7 @@ mod tests { let child = s.append("child", vec![], Some(root)).expect("child"); let grandchild = s.append("gc", vec![], Some(child)).expect("gc"); assert_eq!(s.settle(), vec![root]); - s.complete(root, Outcome::Failed); + s.complete(root, Outcome::Failed(String::new())); assert_eq!(s.graph().node(child).unwrap().state, State::Cancelled); assert_eq!(s.graph().node(grandchild).unwrap().state, State::Cancelled); }