git_policy_cli/
subvolume_journal.rs

1//! What:
2//!  The journal of a subvolume migration: a small JSON file beside the checkout that records which steps finished,
3//!  written atomically after each step.
4//! Why:
5//!  A rerun after a kill at any instant resumes from it: a step marked `done` is skipped, a step marked `started` is
6//!  repeated (every step is idempotent), and a step with no entry has not begun. The file lives in the checkout's
7//!  parent directory (`.cli-git-migrate-<name>.journal.json`), outside the tree that is copied and exchanged, so it
8//!  survives the exchange. It is replaced by writing a sibling file and renaming it, so a kill never leaves half a
9//!  journal. `serde_json` is already a dependency of the crate.
10//!
11//! In TS you'd write (pseudocode):
12//! ```ts
13//! const journal = await loadJournal(path); await journal.mark('copy', 'started'); ... await journal.mark('copy', 'done');
14//! ```
15
16use super::iso_time::{format_iso_milliseconds, now_iso};
17use super::subvolume_error::MigrateError;
18use serde_json::{Map, Value};
19use std::collections::BTreeMap;
20use std::io::Write;
21use std::path::{Path, PathBuf};
22
23/// The journal format version this code writes and reads.
24const SCHEMA: u64 = 1;
25
26/// What:
27///  The records of one migration.
28///  `BTreeMap<String, String>` is a text-to-text dictionary kept in key order (sibling `HashMap` has no order).
29/// Why:
30///   A sorted map writes the same file text for the same records.
31///
32/// In TS you'd write (pseudocode):
33/// ```ts
34/// type Journal = { path: string; started: string; steps: Record<string, { state: string; at: string }> };
35/// ```
36#[derive(Clone, Debug, Eq, PartialEq)]
37pub struct Journal {
38    /// Where the file is (or will be).
39    path: PathBuf,
40    /// When the migration began, as `toISOString` text.
41    started: String,
42    /// Step name to state (`started`, `done` or `rolled-back`).
43    states: BTreeMap<String, String>,
44    /// Step name to the time in milliseconds after the epoch at which it got its current state.
45    times: BTreeMap<String, i64>,
46    /// Whether the file exists on disk.
47    persisted: bool,
48}
49
50impl Journal {
51    /// What:
52    ///  Load the journal at `path`, or start an empty one when there is no file.
53    /// Why:
54    ///   A file that exists but is not a journal of this format is an error naming the file: guessing would resume
55    ///  or skip steps on a false record.
56    ///
57    /// In TS you'd write (pseudocode):
58    /// ```ts
59    /// const journal = await Journal.load(path);
60    /// ```
61    pub fn load(path: &Path) -> Result<Journal, MigrateError> {
62        let text: String = match std::fs::read_to_string(path) {
63            Ok(found) => found,
64            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
65                return Ok(Journal { path: path.to_path_buf(), started: now_iso(), states: BTreeMap::new(), times: BTreeMap::new(), persisted: false });
66            }
67            Err(error) => return Err(MigrateError::io("read the journal", path, &error)),
68        };
69        let broken = |why: &str| { return MigrateError::Journal(format!("the journal {} {why}; read it, and delete it only if no migration is in progress", path.display())); };
70        let value: Value = serde_json::from_str(text.as_str()).map_err(|error: serde_json::Error| { return broken(format!("is not valid JSON ({error})").as_str()); })?;
71        if value.get("schema").and_then(Value::as_u64) != Some(SCHEMA) {
72            return Err(broken("has an unknown schema version"));
73        }
74        let started: String = value.get("started").and_then(Value::as_str).ok_or_else(|| { return broken("has no start time"); })?.to_string();
75        let steps: &Map<String, Value> = value.get("steps").and_then(Value::as_object).ok_or_else(|| { return broken("has no steps"); })?;
76        let mut states: BTreeMap<String, String> = BTreeMap::new();
77        let mut times: BTreeMap<String, i64> = BTreeMap::new();
78        for (name, record) in steps {
79            let state: &str = record.get("state").and_then(Value::as_str).ok_or_else(|| { return broken("has a step without a state"); })?;
80            states.insert(name.clone(), state.to_string());
81            if let Some(milliseconds) = record.get("epoch_ms").and_then(Value::as_i64) {
82                times.insert(name.clone(), milliseconds);
83            }
84        }
85        return Ok(Journal { path: path.to_path_buf(), started, states, times, persisted: true });
86    }
87
88    /// When the migration began, as `toISOString` text; unique to the migration and kept by a resumed run.
89    pub fn started(&self) -> &str {
90        return self.started.as_str();
91    }
92
93    /// The journal file's path.
94    pub fn path(&self) -> &Path {
95        return self.path.as_path();
96    }
97
98    /// Whether any step has been recorded on disk.
99    pub fn is_persisted(&self) -> bool {
100        return self.persisted;
101    }
102
103    /// The time, in whole seconds after the epoch, at which the step got its current state; nothing for a step with no
104    ///  record or a record written without a time.
105    pub fn seconds_of(&self, step: &str) -> Option<i64> {
106        return self.times.get(step).map(|milliseconds: &i64| { return milliseconds.div_euclid(1000); });
107    }
108
109    /// The recorded state of a step, if any.
110    pub fn state(&self, step: &str) -> Option<&str> {
111        return self.states.get(step).map(String::as_str);
112    }
113
114    /// Whether the step finished.
115    pub fn is_done(&self, step: &str) -> bool {
116        return self.state(step) == Some("done");
117    }
118
119    /// Whether the step began (finished or not).
120    pub fn has_begun(&self, step: &str) -> bool {
121        return matches!(self.state(step), Some("started" | "done"));
122    }
123
124    /// What:
125    ///  Record a step's state and write the file.
126    /// Why:
127    ///   The file is written to a sibling and renamed over the journal, with the data synced first, so a kill leaves
128    ///  either the old or the new journal.
129    ///
130    /// In TS you'd write (pseudocode):
131    /// ```ts
132    /// await journal.mark('copy', 'done');
133    /// ```
134    pub fn mark(&mut self, step: &str, state: &str) -> Result<(), MigrateError> {
135        self.states.insert(step.to_string(), state.to_string());
136        let now: i64 = std::time::SystemTime::now()
137            .duration_since(std::time::UNIX_EPOCH)
138            .map_or(0, |elapsed: std::time::Duration| { return i64::try_from(elapsed.as_millis()).unwrap_or(i64::MAX); });
139        self.times.insert(step.to_string(), now);
140        let mut steps: Map<String, Value> = Map::new();
141        for (name, recorded) in &self.states {
142            let milliseconds: i64 = self.times.get(name).copied().unwrap_or(now);
143            let mut record: Map<String, Value> = Map::new();
144            record.insert(String::from("state"), Value::String(recorded.clone()));
145            record.insert(String::from("at"), Value::String(format_iso_milliseconds(milliseconds)));
146            record.insert(String::from("epoch_ms"), Value::from(milliseconds));
147            steps.insert(name.clone(), Value::Object(record));
148        }
149        let mut document: Map<String, Value> = Map::new();
150        document.insert(String::from("schema"), Value::from(SCHEMA));
151        document.insert(String::from("started"), Value::String(self.started.clone()));
152        document.insert(String::from("steps"), Value::Object(steps));
153        let text: String = serde_json::to_string_pretty(&Value::Object(document)).unwrap_or_default();
154        let mut sibling: std::ffi::OsString = self.path.as_os_str().to_os_string();
155        sibling.push(".tmp");
156        let sibling_path: PathBuf = PathBuf::from(sibling);
157        let write = || -> std::io::Result<()> {
158            let mut file: std::fs::File = std::fs::File::create(&sibling_path)?;
159            file.write_all(text.as_bytes())?;
160            file.sync_all()?;
161            return std::fs::rename(&sibling_path, &self.path);
162        };
163        write().map_err(|error: std::io::Error| { return MigrateError::io("write the journal", self.path.as_path(), &error); })?;
164        self.persisted = true;
165        return Ok(());
166    }
167
168    /// What:
169    ///  Delete the journal file.
170    /// Why:
171    ///   A rolled-back or fully removed migration leaves no record, so the next migration starts clean.
172    ///
173    /// In TS you'd write (pseudocode):
174    /// ```ts
175    /// await rm(journal.path, { force: true });
176    /// ```
177    pub fn delete(&mut self) -> Result<(), MigrateError> {
178        match std::fs::remove_file(&self.path) {
179            Ok(()) => {}
180            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
181            Err(error) => return Err(MigrateError::io("delete the journal", self.path.as_path(), &error)),
182        }
183        self.states.clear();
184        self.times.clear();
185        self.persisted = false;
186        return Ok(());
187    }
188}
189
190/// Journal controls stay out of the release executable.
191#[cfg(test)]
192#[path = "subvolume_journal_tests.rs"]
193mod tests;
git_policy_cli/subvolume_journal.rs

git_policy_cli/
subvolume_journal.rs
