isothrum_runner/
store.rs

1//! The runner's state, one folder per run under the embedder's home:
2//!
3//! ```text
4//! <home>/runs/<run name>/
5//!   record.json     the run's record (atomic, owner-only)
6//!   lock            the OS lock: one operation per run across processes
7//!   events.jsonl    the last operations' events, so a reattached UI can replay them
8//!   …               what the manifest keeps in `${run.state_dir}` (Compose's pinned ports)
9//! ```
10//!
11//! The run's folder holds only the run and its generated files, so replacing it never strands a
12//! run. `Store` is a struct over the file system, not a trait: tests use a temporary home.
13
14mod fs;
15mod lock;
16
17use std::collections::BTreeMap;
18use std::io::{BufRead, BufReader};
19use std::path::{Path, PathBuf};
20
21use isoloom_manifest::{Cloud, Driver, Target};
22use serde::{Deserialize, Serialize};
23
24pub(crate) use fs::{create_private, private_dir, write_private};
25pub(crate) use lock::RunLocks;
26
27use crate::Result;
28use crate::error::{Error, ErrorCode};
29use crate::events::Event;
30use crate::name::RunName;
31use crate::state::{Op, RunState};
32
33/// The record format's version.
34const RECORD_VERSION: u32 = 0;
35
36/// How many event lines a run's log keeps before an operation trims it to half.
37const EVENTS_KEPT: usize = 4000;
38
39/// The last operation on a run and how it ended.
40#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
41#[non_exhaustive]
42pub struct LastOp {
43    /// The operation.
44    pub op: Op,
45    /// Whether it succeeded.
46    pub ok: bool,
47    /// Its failure's code, when it failed.
48    #[serde(default, skip_serializing_if = "Option::is_none")]
49    pub code: Option<ErrorCode>,
50    /// When it ended.
51    pub at: String,
52}
53
54/// What the runner keeps about a run (never a secret's value).
55#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
56#[non_exhaustive]
57pub struct RunRecord {
58    /// The record format's version.
59    pub version: u32,
60    /// The run's name.
61    pub name: RunName,
62    /// The manifest the run was started from, absolute.
63    pub manifest: PathBuf,
64    /// Its SHA-256.
65    pub manifest_sha256: String,
66    /// Its target.
67    pub target: Target,
68    /// Its driver.
69    pub driver: Driver,
70    /// Its cloud, for a cloud target's run: what the reaper picks runs by.
71    #[serde(default, skip_serializing_if = "Option::is_none")]
72    pub cloud: Option<Cloud>,
73    /// The Vagrant provider the run uses, when it has one.
74    #[serde(default, skip_serializing_if = "Option::is_none")]
75    pub provider: Option<String>,
76    /// What the embedder calls the place the run is on (a server, an account).
77    #[serde(default, skip_serializing_if = "Option::is_none")]
78    pub host_label: Option<String>,
79    /// Where it stood after its last operation.
80    pub state: RunState,
81    /// Every param the run was started with: a later operation (a `down`) replays them, so it acts
82    /// on what was created, not on what the embedder's settings say now.
83    pub params: BTreeMap<String, String>,
84    /// When it was last started.
85    #[serde(default, skip_serializing_if = "Option::is_none")]
86    pub started_at: Option<String>,
87    /// Its slot on its place (`${run.slot}`, 1 to 99), held from its first start to its `down`.
88    #[serde(default, skip_serializing_if = "Option::is_none")]
89    pub slot: Option<u8>,
90    /// The login its SSH goes through first (`user@host`), when the embedder gave one.
91    #[serde(default, skip_serializing_if = "Option::is_none")]
92    pub ssh_jump: Option<String>,
93    /// When it stops itself (the manifest's `expiry`), in seconds since the Unix epoch.
94    #[serde(default, skip_serializing_if = "Option::is_none")]
95    pub expires_at: Option<u64>,
96    /// The last operation.
97    #[serde(default, skip_serializing_if = "Option::is_none")]
98    pub last: Option<LastOp>,
99}
100
101impl RunRecord {
102    /// A record of a run never started yet.
103    pub(crate) fn fresh(name: RunName, manifest: PathBuf, sha256: String, target: Target, driver: Driver) -> RunRecord {
104        RunRecord {
105            version: RECORD_VERSION,
106            name,
107            manifest,
108            manifest_sha256: sha256,
109            target,
110            driver,
111            cloud: None,
112            provider: None,
113            host_label: None,
114            state: RunState::Absent,
115            params: BTreeMap::new(),
116            started_at: None,
117            slot: None,
118            ssh_jump: None,
119            expires_at: None,
120            last: None,
121        }
122    }
123}
124
125/// The runner's state folder.
126#[derive(Debug, Clone)]
127pub(crate) struct Store {
128    runs: PathBuf,
129}
130
131impl Store {
132    /// The store under `home`, its `runs/` folder created owner-only.
133    pub(crate) fn open(home: &Path) -> Result<Store> {
134        let runs = home.join("runs");
135        private_dir(&runs).map_err(Error::io(&runs, "created"))?;
136        Ok(Store { runs })
137    }
138
139    /// Run `name`'s folder, created owner-only.
140    pub(crate) fn run_dir(&self, name: &RunName) -> Result<PathBuf> {
141        let dir = self.runs.join(name.as_str());
142        private_dir(&dir).map_err(Error::io(&dir, "created"))?;
143        Ok(dir)
144    }
145
146    /// Run `name`'s lock file.
147    pub(crate) fn lock_path(&self, name: &RunName) -> PathBuf {
148        self.runs.join(name.as_str()).join("lock")
149    }
150
151    /// Run `name`'s record, when it has one.
152    pub(crate) fn record(&self, name: &RunName) -> Result<Option<RunRecord>> {
153        let path = self.runs.join(name.as_str()).join("record.json");
154        match std::fs::read(&path) {
155            Ok(bytes) => serde_json::from_slice(&bytes).map(Some).map_err(|e| Error::io(&path, "read")(e.into())),
156            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
157            Err(e) => Err(Error::io(&path, "read")(e)),
158        }
159    }
160
161    /// Writes `record` whole.
162    pub(crate) fn save(&self, record: &RunRecord) -> Result<()> {
163        let path = self.run_dir(&record.name)?.join("record.json");
164        let text = serde_json::to_vec_pretty(record).map_err(|e| Error::io(&path, "written")(e.into()))?;
165        write_private(&path, &text).map_err(Error::io(&path, "written"))
166    }
167
168    /// Every run's record, by name.
169    pub(crate) fn records(&self) -> Result<Vec<RunRecord>> {
170        let entries = std::fs::read_dir(&self.runs).map_err(Error::io(&self.runs, "read"))?;
171        let mut names: Vec<RunName> = entries.flatten().filter_map(|e| RunName::new(&e.file_name().to_string_lossy()).ok()).collect();
172        names.sort();
173        let mut records = Vec::new();
174        for name in names {
175            records.extend(self.record(&name)?);
176        }
177        Ok(records)
178    }
179
180    /// Run `name`'s event log, opened to append, trimmed to its newer half once it is long.
181    pub(crate) fn event_log(&self, name: &RunName) -> Result<std::fs::File> {
182        let path = self.run_dir(name)?.join("events.jsonl");
183        let lines = read_lines(&path)?;
184        if lines.len() > EVENTS_KEPT {
185            let newer: Vec<&str> = lines.iter().skip(lines.len().saturating_sub(EVENTS_KEPT / 2)).map(String::as_str).collect();
186            let kept = newer.join("\n") + "\n";
187            write_private(&path, kept.as_bytes()).map_err(Error::io(&path, "written"))?;
188        }
189        fs::append_private(&path).map_err(Error::io(&path, "written"))
190    }
191
192    /// Run `name`'s logged events, oldest first (a line that doesn't parse is skipped).
193    pub(crate) fn events(&self, name: &RunName) -> Result<Vec<Event>> {
194        let path = self.runs.join(name.as_str()).join("events.jsonl");
195        Ok(read_lines(&path)?.iter().filter_map(|l| serde_json::from_str(l).ok()).collect())
196    }
197}
198
199/// `path`'s lines; none when it doesn't exist.
200fn read_lines(path: &Path) -> Result<Vec<String>> {
201    match std::fs::File::open(path) {
202        Ok(f) => BufReader::new(f).lines().collect::<Result<_, _>>().map_err(Error::io(path, "read")),
203        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
204        Err(e) => Err(Error::io(path, "read")(e)),
205    }
206}
207
208#[cfg(test)]
209mod tests {
210    use std::io::Write;
211
212    use super::*;
213
214    fn record(name: &str) -> RunRecord {
215        RunRecord::fresh(
216            RunName::new(name).unwrap(),
217            "/labs/x/.isoloom/docker/run.json".into(),
218            "ab".repeat(32),
219            Target::Docker,
220            Driver::Compose,
221        )
222    }
223
224    #[test]
225    fn records_are_kept_per_run_and_listed_by_name() {
226        let home = tempfile::tempdir().unwrap();
227        let store = Store::open(home.path()).unwrap();
228        assert_eq!(store.record(&RunName::new("b").unwrap()).unwrap(), None);
229        let mut b = record("b");
230        b.params.insert("ISOLOOM_PUBLISH_FIXED".into(), "1".into());
231        store.save(&b).unwrap();
232        store.save(&record("a")).unwrap();
233        std::fs::create_dir_all(home.path().join("runs").join("Not A Run")).unwrap();
234        let names: Vec<String> = store.records().unwrap().iter().map(|r| r.name.to_string()).collect();
235        assert_eq!(names, ["a", "b"]);
236        assert_eq!(store.record(&b.name).unwrap(), Some(b));
237    }
238
239    #[test]
240    fn the_event_log_is_trimmed_to_its_newer_half() {
241        let home = tempfile::tempdir().unwrap();
242        let store = Store::open(home.path()).unwrap();
243        let name = RunName::new("r").unwrap();
244        let mut log = store.event_log(&name).unwrap();
245        for seq in 0..=EVENTS_KEPT {
246            writeln!(log, "{{\"seq\":{seq}}}").unwrap();
247        }
248        drop(log);
249        drop(store.event_log(&name).unwrap());
250        let lines = read_lines(&home.path().join("runs/r/events.jsonl")).unwrap();
251        assert_eq!(lines.len(), EVENTS_KEPT / 2);
252        assert_eq!(lines.last().unwrap(), &format!("{{\"seq\":{EVENTS_KEPT}}}"));
253    }
254}
isothrum_runner/store.rs

isothrum_runner/
store.rs
