diff --git a/jouleclaw/SPEC.md b/jouleclaw/SPEC.md index ddf4b270a..47db2db4d 100644 --- a/jouleclaw/SPEC.md +++ b/jouleclaw/SPEC.md @@ -942,6 +942,45 @@ Reference: [`crates/jouleclaw-graph`](./jouleclaw-rs/crates/jouleclaw-graph/) `jc:WorkerHandoffShape` at [`vocab/v1.shapes.ttl`](./vocab/v1.shapes.ttl). +### 17.3 A node in flight survives the process + +The run journal records a node when it completes. A node whose +process dies before it completes would otherwise leave no trace, +and a restarted run would dispatch it again as if it had never +begun, repeating any work it did outside the run: a message sent, +a payment made, a record written. + +A consumer that dispatches a node SHOULD therefore record the +dispatch (`RunEvent::NodeDispatched`) **before** the node's work +begins, with the node's **replay rule**: + +- `safe`: running the node again is harmless, because it is a read + or its work is keyed so a repeat finds its first result. +- `never` (the default): the node may have acted outside the run, + and nothing in the journal says whether it did. + +A dispatched node that is not yet recorded is **in flight**, and +MUST NOT be dispatched again. After a restart, the consumer MUST +mark every node still in flight as interrupted +(`RunEvent::NodeInterrupted`) before dispatching anything, keeping +the energy observed for the attempt as a lower bound on what it +spent. An interrupted `safe` node is ready again, as a new attempt. +An interrupted `never` node is **held**: whether its work took +effect is not knowable from the journal, so it MUST NOT run again +until it is settled, by recording what it did, by an approval to +run it again (`RunEvent::RerunApproved`) that names who gave it, +or by failing the run. It is not left to the model to decide: a +model told only that a call "may have partly run" can repeat a +write it cannot see. + +Until a run is recovered, a node in flight stays out of the ready +set, so a consumer that omits recovery stalls rather than +repeating work. This is the run-journal form of JCP SPEC §4.11. + +Reference: `Run::dispatch`, `Run::recover`, `Run::approve_rerun`, +`Replay`, `NodeAttempt` in +[`crates/jouleclaw-graph`](./jouleclaw-rs/crates/jouleclaw-graph/). + --- ## 18. Mission orchestrator diff --git a/jouleclaw/jouleclaw-rs/crates/jouleclaw-graph/src/lib.rs b/jouleclaw/jouleclaw-rs/crates/jouleclaw-graph/src/lib.rs index 793c332ee..33ed41f3e 100644 --- a/jouleclaw/jouleclaw-rs/crates/jouleclaw-graph/src/lib.rs +++ b/jouleclaw/jouleclaw-rs/crates/jouleclaw-graph/src/lib.rs @@ -21,6 +21,15 @@ //! graph; that is a separate concern from "did this step finish." //! The brief excluded time-travel replay in v1; this is its //! structural enforcement. +//! - **A node in flight survives the process.** A consumer that +//! records a node's dispatch ([`Run::dispatch`]) before running it +//! leaves the node on record as in flight, so a process that dies in +//! the middle of it does not resume as if the node had never begun. +//! [`Run::recover`] marks such a node interrupted. A node dispatched as +//! [`Replay::Safe`] is then ready again; one dispatched as +//! [`Replay::Never`] (the default) is held until the consumer settles +//! it, because whether its work took effect is not in the journal +//! (SPEC §17.3). //! - **No human-in-the-loop checkpoints.** A pending HITL approval is //! modelled by the consumer holding a node un-recorded; the graph //! doesn't enforce a "waiting for approval" state in v1. @@ -279,6 +288,49 @@ fn reachable_from(g: &RunGraph, start: NodeId) -> HashSet { // Journal — append-only event log with blake3 hash chain // ───────────────────────────────────────────────────────────────────── +/// Whether a node may run again after a crash interrupted it (SPEC §17.3). +/// Declared when the node is dispatched, and recorded with the dispatch. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Replay { + /// Running the node again is harmless: a read, or work keyed so a + /// repeat finds its first result. After a crash it is ready again, as a + /// new attempt. + Safe, + /// The node may have acted outside the run (sent a message, moved money, + /// written to another system), and nothing in the journal says whether it + /// did. After a crash it is held for the consumer to settle: record what + /// happened, approve a rerun, or fail the run. The default. + #[default] + Never, +} + +/// Where a dispatched node stands (SPEC §17.3), derived from the journal. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum NodeAttempt { + /// Dispatched and not yet recorded: running, or, after a restart, not yet + /// found by [`Run::recover`]. + InFlight { attempt: u32, replay: Replay }, + /// In flight when its process stopped. Whether its work took effect is + /// not knowable from the journal. `observed_uj` is the energy observed for + /// the attempt before it stopped: a lower bound on what it spent. + Interrupted { attempt: u32, replay: Replay, observed_uj: u64, rerun_approved: bool }, +} + +impl NodeAttempt { + /// Whether the node may be dispatched again: interrupted, and either safe + /// to rerun or approved for one. + pub fn may_rerun(&self) -> bool { + matches!(self, NodeAttempt::Interrupted { replay: Replay::Safe, .. } | NodeAttempt::Interrupted { rerun_approved: true, .. }) + } + + fn attempt(&self) -> u32 { + match self { + NodeAttempt::InFlight { attempt, .. } | NodeAttempt::Interrupted { attempt, .. } => *attempt, + } + } +} + /// One event in a [`Journal`]. The variant set is exhaustive and /// versioned by the `jc_journal` schema; new variants are additive /// and must serialise to a JSON discriminant that older readers @@ -299,6 +351,31 @@ pub enum RunEvent { output_hash: String, recorded_at: u64, }, + /// A node's work is about to begin. Recorded before it begins, so a node + /// the process dies in is on record as in flight rather than absent + /// (SPEC §17.3). `attempt` counts from 1. + NodeDispatched { + node_id: NodeId, + attempt: u32, + replay: Replay, + dispatched_at: u64, + }, + /// A node in flight when its process stopped, found by [`Run::recover`]. + /// Its attempt may or may not have taken effect; `observed_uj` is the + /// energy observed for it before it stopped, a lower bound on its spend. + NodeInterrupted { + node_id: NodeId, + attempt: u32, + observed_uj: u64, + recovered_at: u64, + }, + /// The consumer approved running an interrupted [`Replay::Never`] node + /// again, and says who approved it. + RerunApproved { + node_id: NodeId, + approved_by: String, + approved_at: u64, + }, /// An explicit energy observation that does NOT complete a node. /// Useful when a per-token attributor, an external meter, or a /// retry attempt produces an energy reading the runtime should @@ -761,6 +838,16 @@ pub enum RunError { AlreadyRecorded(NodeId), #[error("run is no longer in progress (status: {0:?})")] NotInProgress(RunStatus), + /// The node is dispatched and not yet recorded. + #[error("node {0} is already in flight")] + InFlight(NodeId), + /// The node was interrupted and is held: it may have taken effect, so it + /// is settled before it runs again (SPEC §17.3). + #[error("node {0} was interrupted and is held until it is settled")] + Held(NodeId), + /// A rerun was approved for a node that is not held. + #[error("node {0} is not an interrupted node awaiting settlement")] + NotHeld(NodeId), } impl Run { @@ -790,14 +877,20 @@ impl Run { /// Nodes that are now ready to dispatch — present in the graph, not /// yet completed, and whose merge strategy is satisfied. The entry - /// node is ready until it is recorded. + /// node is ready until it is recorded. A node in flight is not ready, + /// so it is not dispatched twice, and neither is an interrupted node + /// until it may run again ([`NodeAttempt::may_rerun`]). pub fn ready_nodes(&self) -> Vec { let done = self.completed_nodes(); + let attempts = self.attempts(); let mut out = Vec::new(); for n in &self.graph.nodes { if done.contains(&n.id) { continue; } + if attempts.get(&n.id).is_some_and(|a| !a.may_rerun()) { + continue; + } let preds = self.graph.predecessors(n.id); let ready = if preds.is_empty() { // No predecessors → only the entry is ready by @@ -832,7 +925,10 @@ impl Run { if self.completed_nodes().contains(&checkpoint.node_id) { return Err(RunError::AlreadyRecorded(checkpoint.node_id)); } - if !self.ready_nodes().contains(&checkpoint.node_id) { + // A dispatched node records its result; an interrupted one is settled + // by recording what happened. Either was ready when it was dispatched. + let dispatched = self.attempt_state(checkpoint.node_id).is_some(); + if !dispatched && !self.ready_nodes().contains(&checkpoint.node_id) { return Err(RunError::NotReady(checkpoint.node_id)); } let event = RunEvent::NodeRecorded { @@ -853,6 +949,128 @@ impl Run { Ok(()) } + /// Record that a ready node's work is about to begin, and how it may be + /// treated if a crash interrupts it (SPEC §17.3). Call it before running + /// the node: the record is what lets a restarted process tell a node that + /// began from one that never did. Returns the attempt number, counting + /// from 1. + pub fn dispatch(&mut self, node_id: NodeId, replay: Replay, dispatched_at: u64) -> Result { + if self.status != RunStatus::InProgress { + return Err(RunError::NotInProgress(self.status)); + } + if self.graph.node(node_id).is_none() { + return Err(RunError::UnknownNode(node_id)); + } + if self.completed_nodes().contains(&node_id) { + return Err(RunError::AlreadyRecorded(node_id)); + } + let prior = self.attempt_state(node_id); + match prior { + Some(NodeAttempt::InFlight { .. }) => return Err(RunError::InFlight(node_id)), + Some(a @ NodeAttempt::Interrupted { .. }) if !a.may_rerun() => return Err(RunError::Held(node_id)), + _ => {} + } + if !self.ready_nodes().contains(&node_id) { + return Err(RunError::NotReady(node_id)); + } + let attempt = prior.map_or(1, |a| a.attempt() + 1); + self.journal.append(RunEvent::NodeDispatched { node_id, attempt, replay, dispatched_at }); + Ok(attempt) + } + + /// After a restart: mark every node still in flight as interrupted, + /// since the process that ran it is gone (SPEC §17.3). Call it once a + /// loaded or replayed run is in the new process, before dispatching + /// anything. Returns the nodes it marked. A node dispatched as + /// [`Replay::Safe`] is ready again; one dispatched as [`Replay::Never`] + /// is held until it is settled. + pub fn recover(&mut self, recovered_at: u64) -> Vec { + let mut marked = Vec::new(); + if self.status != RunStatus::InProgress { + return marked; + } + let observed = self.observed_since_dispatch(); + for (node_id, a) in self.attempts() { + if let NodeAttempt::InFlight { attempt, .. } = a { + let observed_uj = observed.get(&node_id).copied().unwrap_or(0); + self.journal.append(RunEvent::NodeInterrupted { node_id, attempt, observed_uj, recovered_at }); + marked.push(node_id); + } + } + marked + } + + /// Settle a held node by running it again: whoever decides records + /// that they did, so the rerun is attributable (SPEC §17.3). + pub fn approve_rerun(&mut self, node_id: NodeId, approved_by: impl Into, approved_at: u64) -> Result<(), RunError> { + match self.attempt_state(node_id) { + Some(NodeAttempt::Interrupted { replay: Replay::Never, rerun_approved: false, .. }) => { + self.journal.append(RunEvent::RerunApproved { node_id, approved_by: approved_by.into(), approved_at }); + Ok(()) + } + _ => Err(RunError::NotHeld(node_id)), + } + } + + /// Where a dispatched, unrecorded node stands, if it is one. + pub fn attempt_state(&self, node_id: NodeId) -> Option { + self.attempts().remove(&node_id) + } + + /// Nodes interrupted by a stopped process and held until settled. + pub fn held_nodes(&self) -> Vec<(NodeId, NodeAttempt)> { + self.attempts().into_iter().filter(|(_, a)| matches!(a, NodeAttempt::Interrupted { .. }) && !a.may_rerun()).collect() + } + + /// The lifecycle of every dispatched node not yet recorded, by + /// replaying the dispatch events in the journal. + fn attempts(&self) -> std::collections::BTreeMap { + let mut out = std::collections::BTreeMap::new(); + for entry in &self.journal.entries { + match &entry.event { + RunEvent::NodeDispatched { node_id, attempt, replay, .. } => { + out.insert(*node_id, NodeAttempt::InFlight { attempt: *attempt, replay: *replay }); + } + RunEvent::NodeRecorded { node_id, .. } => { + out.remove(node_id); + } + RunEvent::NodeInterrupted { node_id, attempt, observed_uj, .. } => { + let replay = match out.get(node_id) { + Some(NodeAttempt::InFlight { replay, .. }) => *replay, + _ => Replay::Never, + }; + out.insert(*node_id, NodeAttempt::Interrupted { attempt: *attempt, replay, observed_uj: *observed_uj, rerun_approved: false }); + } + RunEvent::RerunApproved { node_id, .. } => { + if let Some(NodeAttempt::Interrupted { rerun_approved, .. }) = out.get_mut(node_id) { + *rerun_approved = true; + } + } + _ => {} + } + } + out + } + + /// Energy observed for each node since its latest dispatch. + fn observed_since_dispatch(&self) -> std::collections::BTreeMap { + let mut out = std::collections::BTreeMap::new(); + for entry in &self.journal.entries { + match &entry.event { + RunEvent::NodeDispatched { node_id, .. } => { + out.insert(*node_id, 0u64); + } + RunEvent::EnergyObserved { node_id, observed_uj, .. } => { + if let Some(total) = out.get_mut(node_id) { + *total = total.saturating_add(*observed_uj); + } + } + _ => {} + } + } + out + } + /// Mark the run as failed (consumer decides when; e.g. a verifier /// rejected a critical node's output). Appends a /// [`RunEvent::Failed`] to the journal. @@ -989,6 +1207,9 @@ impl Run { RunEvent::EnergyObserved { .. } => {} RunEvent::HandoffRecorded { .. } => {} RunEvent::FeedbackReported { .. } => {} + // The dispatch lifecycle is derived from the journal on demand + // (`Run::attempts`), so replay carries it without rebuilding it. + RunEvent::NodeDispatched { .. } | RunEvent::NodeInterrupted { .. } | RunEvent::RerunApproved { .. } => {} } } Ok(run) @@ -1522,6 +1743,120 @@ mod tests { p } + /// A two-node run, nothing recorded: send (1), then report (2). + fn two(id: &str) -> Run { + let g = RunGraph { + entry: NodeId(1), + terminals: vec![NodeId(2)], + nodes: vec![n(1, "send"), n(2, "report")], + edges: vec![edge(1, 2)], + }; + Run::start(g, id).unwrap() + } + + /// The process dies: a new one has only the journal. + fn restart(run: &Run) -> Run { + Run::replay_from(run.graph.clone(), &run.journal.entries, run.run_id.clone()).unwrap() + } + + /// A node the process died in is not run again as if it had never + /// begun. Without a dispatch record it used to be ready again after a + /// restart, so the next dispatch repeated its work silently. Dispatched + /// as `Never`, it is held: its energy so far is kept, and running it + /// again takes an attributable approval. + #[test] + fn an_interrupted_node_is_held_not_rerun() { + let mut run = two("held"); + assert_eq!(run.dispatch(NodeId(1), Replay::Never, 10).unwrap(), 1); + run.observe_energy(NodeId(1), 40, Provenance::HwShunt); + + let mut after = restart(&run); + assert!(!after.ready_nodes().contains(&NodeId(1)), "in flight, so not ready even before recovery"); + assert_eq!(after.recover(20), vec![NodeId(1)]); + assert_eq!( + after.attempt_state(NodeId(1)), + Some(NodeAttempt::Interrupted { attempt: 1, replay: Replay::Never, observed_uj: 40, rerun_approved: false }) + ); + assert!(after.ready_nodes().is_empty()); + assert_eq!(after.dispatch(NodeId(1), Replay::Never, 21), Err(RunError::Held(NodeId(1)))); + assert_eq!(after.held_nodes().len(), 1); + + after.approve_rerun(NodeId(1), "did:key:operator", 30).unwrap(); + assert_eq!(after.ready_nodes(), vec![NodeId(1)]); + assert_eq!(after.dispatch(NodeId(1), Replay::Never, 31).unwrap(), 2); + after.record(ck(1, 90, Provenance::HwShunt)).unwrap(); + assert_eq!(after.ready_nodes(), vec![NodeId(2)]); + assert!(after.held_nodes().is_empty()); + } + + /// A node dispatched as safe to rerun is ready again after a crash, as a + /// new attempt, and needs no approval. + #[test] + fn a_safe_node_is_ready_again_after_a_crash() { + let mut run = two("safe"); + run.dispatch(NodeId(1), Replay::Safe, 10).unwrap(); + let mut after = restart(&run); + after.recover(20); + assert_eq!(after.ready_nodes(), vec![NodeId(1)]); + assert_eq!(after.approve_rerun(NodeId(1), "anyone", 21), Err(RunError::NotHeld(NodeId(1)))); + assert_eq!(after.dispatch(NodeId(1), Replay::Safe, 22).unwrap(), 2); + } + + /// A node in flight is not ready, so it is not dispatched twice. + #[test] + fn an_in_flight_node_is_not_dispatched_twice() { + let mut run = two("twice"); + run.dispatch(NodeId(1), Replay::Safe, 10).unwrap(); + assert!(run.ready_nodes().is_empty()); + assert_eq!(run.dispatch(NodeId(1), Replay::Safe, 11), Err(RunError::InFlight(NodeId(1)))); + run.record(ck(1, 5, Provenance::HwShunt)).unwrap(); + assert_eq!(run.ready_nodes(), vec![NodeId(2)]); + assert_eq!(run.attempt_state(NodeId(1)), None); + } + + /// Recording what an interrupted node did settles it. + #[test] + fn recording_settles_an_interrupted_node() { + let mut run = two("settle"); + run.dispatch(NodeId(1), Replay::Never, 10).unwrap(); + let mut after = restart(&run); + after.recover(20); + after.record(ck(1, 70, Provenance::Estimator)).unwrap(); + assert_eq!(after.ready_nodes(), vec![NodeId(2)]); + assert!(after.held_nodes().is_empty()); + } + + /// Loading a run through a store is a restart too. Until it is + /// recovered, the node stays in flight and out of `ready_nodes`, so a + /// consumer that forgets `recover` stalls rather than repeating work. + #[test] + fn a_run_loaded_from_a_store_keeps_its_node_in_flight() { + let dir = tmpdir("in-flight"); + let mut store = FileRunStore::open(&dir).unwrap(); + let mut run = two("stored"); + run.dispatch(NodeId(1), Replay::Never, 10).unwrap(); + store.save(&run); + let mut loaded = store.load("stored").unwrap(); + assert!(loaded.ready_nodes().is_empty()); + assert_eq!(loaded.recover(20), vec![NodeId(1)]); + assert_eq!(loaded.held_nodes().len(), 1); + let _ = std::fs::remove_dir_all(&dir); + } + + /// The dispatch lifecycle is in the hash chain and replays exactly. + #[test] + fn dispatch_events_are_chained_and_replay_exactly() { + let mut run = two("chain"); + run.dispatch(NodeId(1), Replay::Never, 10).unwrap(); + let mut after = restart(&run); + after.recover(20); + after.approve_rerun(NodeId(1), "did:key:operator", 30).unwrap(); + assert!(after.journal.verify_chain().is_ok()); + let again = restart(&after); + assert_eq!(again.journal, after.journal); + assert_eq!(again.attempt_state(NodeId(1)), after.attempt_state(NodeId(1))); + } + fn linear_run(id: &str) -> Run { let g = RunGraph { entry: NodeId(1), diff --git a/jouleclaw/jouleclaw-rs/crates/jouleclaw-stuck/src/lib.rs b/jouleclaw/jouleclaw-rs/crates/jouleclaw-stuck/src/lib.rs index fbcedc6db..7d84c61a6 100644 --- a/jouleclaw/jouleclaw-rs/crates/jouleclaw-stuck/src/lib.rs +++ b/jouleclaw/jouleclaw-rs/crates/jouleclaw-stuck/src/lib.rs @@ -511,6 +511,12 @@ fn last_observed_timestamp(entries: &[JournalEntry]) -> Option { let ts = match &e.event { RunEvent::NodeRecorded { recorded_at, .. } => Some(*recorded_at), RunEvent::HandoffRecorded { closed_at, .. } => Some(*closed_at), + // A node beginning, found interrupted, or approved for a rerun is + // activity too: a run whose last event is a long node's dispatch + // is working, not abandoned. + RunEvent::NodeDispatched { dispatched_at, .. } => Some(*dispatched_at), + RunEvent::NodeInterrupted { recovered_at, .. } => Some(*recovered_at), + RunEvent::RerunApproved { approved_at, .. } => Some(*approved_at), _ => None, }; if let Some(t) = ts { @@ -705,6 +711,24 @@ mod tests { assert_eq!(sigs.len(), 1); } + /// A node's dispatch is activity: a run whose last event is a long + /// node beginning is working, not abandoned. + #[test] + fn a_recent_dispatch_is_activity() { + let mut r = fresh_run("dispatching"); + r.record_handoff(WorkerHandoffEvent { + worker: "w".into(), + closed_at: 1_700_000_000, + ..Default::default() + }); + r.dispatch(NodeId(1), jouleclaw_graph::Replay::Never, 1_700_000_480).unwrap(); + let d = AbandonmentDetector { + idle_grace_secs: 60, + now_secs: 1_700_000_500, + }; + assert!(d.detect(&r.journal).is_empty(), "dispatched 20 s ago"); + } + #[test] fn abandonment_quiet_inside_grace_period() { let mut r = fresh_run("fresh");