diff --git a/src/store/file/document.rs b/src/store/file/document.rs index 569f6701..6d0fa605 100644 --- a/src/store/file/document.rs +++ b/src/store/file/document.rs @@ -242,6 +242,13 @@ pub fn new_run_record(id: &str, workflow_id: &str, started_at: u64) -> RunRecord inputs: serde_json::Map::new(), trigger: None, origin: None, + // Stamped by the caller through `RunRecord::with_executor`. Left unset + // here so this factory stays usable in tests and tools that are not + // actually executing anything; an unstamped record is treated as + // unowned, which is the honest reading. + executor: None, + // Nobody has asked a run to stop before it has begun. + cancel_requested: false, // Both are evidence about a run that has ended, so a run that has only // just started has neither. They are filled in when it settles. summary: None, diff --git a/src/store/file/store_impl.rs b/src/store/file/store_impl.rs index a086729d..00124110 100644 --- a/src/store/file/store_impl.rs +++ b/src/store/file/store_impl.rs @@ -179,6 +179,31 @@ impl FileWorkflowStore { .join(format!("{}.json", safe_component(run_id)?))) } + /// Every parsable run record in the scope's runs directory, unordered. + /// + /// A missing directory is an empty listing, not an error: a scope that has + /// never run anything has nothing to read. A record this build cannot parse + /// is skipped rather than failing the whole listing — history is + /// diagnostic, and one corrupt file should not hide the rest of it. + pub(super) fn read_run_dir(&self) -> Result, WorkflowError> { + let entries = match std::fs::read_dir(&self.runs_dir) { + Ok(entries) => entries, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(source) => { + return Err(WorkflowError::Io { + path: self.runs_dir.clone(), + source, + }); + } + }; + Ok(entries + .filter_map(|entry| entry.ok().map(|e| e.path())) + .filter(|path| is_json(path)) + .filter_map(|path| std::fs::read(&path).ok()) + .filter_map(|body| serde_json::from_slice::(&body).ok()) + .collect()) + } + /// Run one workflow definition mutation while holding its filesystem lock. pub(super) fn with_definition_lock( &self, diff --git a/src/store/file/workflow_store_impl.rs b/src/store/file/workflow_store_impl.rs index e4ff4819..8abf0799 100644 --- a/src/store/file/workflow_store_impl.rs +++ b/src/store/file/workflow_store_impl.rs @@ -135,31 +135,25 @@ impl WorkflowStore for FileWorkflowStore { } fn list_runs(&self, workflow_id: &str) -> Result, WorkflowError> { - let entries = match std::fs::read_dir(&self.runs_dir) { - Ok(entries) => entries, - Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), - Err(source) => { - return Err(WorkflowError::Io { - path: self.runs_dir.clone(), - source, - }); - } - }; - - let mut runs: Vec = entries - .filter_map(|entry| entry.ok().map(|e| e.path())) - .filter(|path| is_json(path)) - // A run record this host cannot parse is skipped rather than - // failing the listing: history is diagnostic, and one corrupt file - // should not hide the rest of it. - .filter_map(|path| std::fs::read(&path).ok()) - .filter_map(|body| serde_json::from_slice::(&body).ok()) + let mut runs: Vec = self + .read_run_dir()? + .into_iter() .filter(|run| run.workflow_id == workflow_id) .collect(); runs.sort_by_key(|run| std::cmp::Reverse(run.started_at)); Ok(runs) } + fn unsettled_runs(&self) -> Result, WorkflowError> { + let mut runs: Vec = self + .read_run_dir()? + .into_iter() + .filter(|run| !run.status.is_settled()) + .collect(); + runs.sort_by_key(|run| std::cmp::Reverse(run.started_at)); + Ok(runs) + } + fn list_revisions(&self, workflow_id: &str) -> Result, WorkflowError> { // Releases before the source/state split kept undo snapshots beside // definitions. Merge that history with new workspace-scoped snapshots diff --git a/src/store/mod.rs b/src/store/mod.rs index 5bfd9d5f..cf9d3778 100644 --- a/src/store/mod.rs +++ b/src/store/mod.rs @@ -40,9 +40,9 @@ pub use file::{safe_component, workspace_scope, write_atomic}; pub use self::types::{ Diagnosis, LEGACY_TRUNCATED_KEY, NoteId, NoteKind, NoteSource, ProposalId, ProposalStatus, - ProposalVerification, RunId, RunOrigin, RunRecord, RunStatus, RunStep, TRUNCATED_KEY, - TranscriptEntry, WorkflowDefaults, WorkflowError, WorkflowId, WorkflowNote, WorkflowProposal, - WorkflowRecord, WorkflowRevision, WorkflowSummary, is_truncated, + ProposalVerification, RunExecutor, RunId, RunOrigin, RunRecord, RunStatus, RunStep, + TRUNCATED_KEY, TranscriptEntry, WorkflowDefaults, WorkflowError, WorkflowId, WorkflowNote, + WorkflowProposal, WorkflowRecord, WorkflowRevision, WorkflowSummary, is_truncated, }; /// An exclusive claim over proposal decisions for one workflow. @@ -146,6 +146,20 @@ pub trait WorkflowStore: Send + Sync { /// Every recorded run for a workflow, newest first. fn list_runs(&self, workflow_id: &str) -> Result, WorkflowError>; + /// Every recorded run that has not settled, across all workflows. + /// + /// What a reconciliation sweep reads — the caller that asks "which records + /// still claim to be live" is asking across every workflow in the scope, and + /// answering that through [`list_runs`](Self::list_runs) would re-read the + /// whole runs directory once per workflow. + /// + /// The default returns nothing, which makes a store that cannot enumerate + /// its runs a store a sweep leaves alone rather than one that fails to + /// compile. + fn unsettled_runs(&self) -> Result, WorkflowError> { + Ok(Vec::new()) + } + /// Every superseded copy of a workflow, newest first. /// /// A workflow that has never been written over has no revisions, which is diff --git a/src/store/tests/mod.rs b/src/store/tests/mod.rs index 2cfb5902..5a0197f4 100644 --- a/src/store/tests/mod.rs +++ b/src/store/tests/mod.rs @@ -31,7 +31,9 @@ pub(super) use super::file::{ pub(super) use super::{ FileWorkflowStore, WorkflowStore, require, require_run, rollback, undo_last, }; -pub(super) use crate::store::types::{RunStatus, WorkflowDefaults, WorkflowError, WorkflowRecord}; +pub(super) use crate::store::types::{ + RunExecutor, RunRecord, RunStatus, WorkflowDefaults, WorkflowError, WorkflowRecord, +}; /// A store rooted in a temporary directory, with definitions and runs kept /// apart the way the discovered layout keeps them. diff --git a/src/store/tests/runs.rs b/src/store/tests/runs.rs index cc914fe8..79e867e6 100644 --- a/src/store/tests/runs.rs +++ b/src/store/tests/runs.rs @@ -49,3 +49,84 @@ fn asking_for_a_run_that_was_never_recorded_is_an_error_not_a_silent_none() { let err = require_run(&store_in(root.path()), "ghost").expect_err("no such run"); assert!(matches!(err, WorkflowError::RunNotFound(_)), "got {err:?}"); } + +#[test] +fn unsettled_runs_spans_every_workflow_and_skips_finished_ones() { + // What a reconciliation sweep reads. It has to cross workflow boundaries — + // "which records still claim to be live" is a question about the whole + // scope — and it must not hand back runs that already have an outcome. + let root = tempfile::tempdir().unwrap(); + let store = store_in(root.path()); + + store + .record_run(&new_run_record("live-alpha", "alpha", 100)) + .unwrap(); + store + .record_run(&new_run_record("live-beta", "beta", 300)) + .unwrap(); + let mut parked = new_run_record("parked", "alpha", 200); + parked.status = RunStatus::PendingApproval; + store.record_run(&parked).unwrap(); + for (id, status) in [ + ("done", RunStatus::Succeeded), + ("broke", RunStatus::Failed), + ("stopped", RunStatus::Cancelled), + ("cut-off", RunStatus::Interrupted), + ] { + let mut record = new_run_record(id, "alpha", 50); + record.status = status; + store.record_run(&record).unwrap(); + } + + let unsettled = store.unsettled_runs().unwrap(); + let ids: Vec<&str> = unsettled.iter().map(|r| r.id.as_str()).collect(); + + assert_eq!( + ids, + vec!["live-beta", "parked", "live-alpha"], + "newest first, both workflows, nothing settled" + ); +} + +#[test] +fn a_scope_that_has_never_run_anything_has_no_unsettled_runs() { + let root = tempfile::tempdir().unwrap(); + let store = store_in(root.path()); + + assert!(store.unsettled_runs().unwrap().is_empty()); +} + +#[test] +fn an_executor_and_a_cancel_request_survive_a_write_and_read() { + let root = tempfile::tempdir().unwrap(); + let store = store_in(root.path()); + let mut record = new_run_record("owned", "alpha", 100).with_executor(Some(RunExecutor { + host: "somewhere".to_string(), + pid: 4711, + started_at_secs: Some(1_700_000_000), + })); + record.cancel_requested = true; + + store.record_run(&record).unwrap(); + + let read = store.get_run("owned").unwrap().unwrap(); + assert_eq!(read.executor, record.executor); + assert!(read.cancel_requested); +} + +#[test] +fn a_record_written_before_executors_existed_still_parses() { + // Both fields are additive, so an older run file must keep loading — and + // must read as unowned, which is what makes a sweep treat it as an orphan. + let older = serde_json::json!({ + "id": "legacy", + "workflowId": "alpha", + "status": "running", + "startedAt": 100, + }); + + let record: RunRecord = serde_json::from_value(older).expect("older records still parse"); + + assert!(record.executor.is_none()); + assert!(!record.cancel_requested); +} diff --git a/src/store/types/mod.rs b/src/store/types/mod.rs index 47c53ea4..6bfb16f5 100644 --- a/src/store/types/mod.rs +++ b/src/store/types/mod.rs @@ -46,8 +46,8 @@ pub use proposal::{ }; pub use run::{ - LEGACY_TRUNCATED_KEY, RunId, RunOrigin, RunRecord, RunStatus, RunStep, TRUNCATED_KEY, - bounded_evidence, bounded_within, is_truncated, + LEGACY_TRUNCATED_KEY, RunExecutor, RunId, RunOrigin, RunRecord, RunStatus, RunStep, + TRUNCATED_KEY, bounded_evidence, bounded_within, is_truncated, }; pub use transcript::TranscriptEntry; pub use workflow::{ diff --git a/src/store/types/run.rs b/src/store/types/run.rs index fdf46ae7..f12a31b4 100644 --- a/src/store/types/run.rs +++ b/src/store/types/run.rs @@ -109,6 +109,39 @@ impl RunOrigin { } } +/// The OS process that is executing a run. +/// +/// A run record on its own cannot say whether the process that wrote it is +/// still alive, which is the difference between "another host is working on +/// this" and "this row is a tombstone from a process that was killed". Stamping +/// the executor turns that unanswerable question into a liveness check. +/// +/// Both halves of the identity matter. A bare pid is not enough: pids are +/// recycled, so a run left behind by pid 4711 would look alive the moment +/// something unrelated is assigned 4711. `started_at_secs` — the executing +/// process's own start time — makes the pair unique for as long as the record +/// is worth reading. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RunExecutor { + /// The host the process runs on. + /// + /// The run store is per-machine today, but a record naming a pid with no + /// host is one shared filesystem away from a liveness check that compares + /// this machine's process table against another machine's pid. Recording it + /// costs nothing and makes the check refuse rather than guess. + pub host: String, + /// The executing process's id. + pub pid: u32, + /// When that process itself started, in seconds since the epoch. + /// + /// The pid-reuse guard. Absent when the platform would not report it, in + /// which case a liveness check falls back to the pid alone and errs toward + /// leaving the record alone. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub started_at_secs: Option, +} + /// Where a run got to. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -220,6 +253,28 @@ pub struct RunRecord { /// caller could not say anything about itself. #[serde(default, skip_serializing_if = "Option::is_none")] pub origin: Option, + /// The process executing this run, while one is. + /// + /// Written when the run is admitted and left in place when it settles: a + /// finished run's executor is history, not a claim. What reads it is a + /// reconciliation sweep, which only consults it for a record that still + /// says [`RunStatus::Running`]. + /// + /// Absent on records written before this field existed, which a sweep + /// treats as unowned — an old record claiming to run is exactly the kind + /// this exists to clean up. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub executor: Option, + /// Whether someone has asked this run to stop. + /// + /// The durable half of cancellation. An in-memory cancellation registry can + /// only reach a run its own process is executing, so a cancel aimed at a run + /// owned by another live process is written here instead; that process + /// notices the flag on its next poll and cancels itself. Never cleared — a + /// run that was asked to stop and then settled should still say it was + /// asked. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub cancel_requested: bool, /// Steps in completion order. #[serde(default)] pub steps: Vec, @@ -283,6 +338,12 @@ impl RunRecord { self } + /// Record the process executing this run. + pub fn with_executor(mut self, executor: Option) -> Self { + self.executor = executor; + self + } + /// How long the run took, once it has settled. pub fn duration_ms(&self) -> Option { self.finished_at diff --git a/src/store/types/types_tests.rs b/src/store/types/types_tests.rs index a337ee22..33f0d8cf 100644 --- a/src/store/types/types_tests.rs +++ b/src/store/types/types_tests.rs @@ -93,6 +93,8 @@ fn run_records_use_camel_case_on_the_wire() { inputs: Default::default(), trigger: None, origin: None, + executor: None, + cancel_requested: false, summary: None, diagnosis: None, }) @@ -175,6 +177,8 @@ fn run_evidence_is_omitted_from_the_wire_when_absent() { inputs: Default::default(), trigger: None, origin: None, + executor: None, + cancel_requested: false, summary: None, diagnosis: None, }) @@ -201,6 +205,8 @@ fn what_a_run_was_started_with_survives_the_wire() { inputs: Default::default(), trigger: None, origin: None, + executor: None, + cancel_requested: false, summary: None, diagnosis: None, }