loftspec_store/
journal.rs

1//! A save through a journal beside `spec/`, the accepted decision "A save is journaled and
2//! finished by the next command": the changed files are staged in `.loft/save/`, renaming the
3//! journal's plan into place commits the save, and each staged file is then renamed into place.
4//! A save interrupted before the commit leaves every file as it was; one interrupted after is
5//! finished by [`recover`], which the file store runs when it is opened.
6
7use std::collections::{BTreeMap, BTreeSet};
8use std::fs;
9use std::io::{self, Write};
10use std::path::{Path, PathBuf};
11
12use loftspec::codec::{Files, SpecPath};
13use serde::{Deserialize, Serialize};
14
15use crate::StoreError;
16use crate::error::io_error;
17use crate::fs::SPEC;
18
19/// The journal's directory under the root.
20pub const JOURNAL: &str = ".loft/save";
21
22const PLAN: &str = "plan.json";
23const PLAN_STAGED: &str = "plan.json.tmp";
24const STAGED: &str = "files";
25const FORMAT: &str = "loftspec-save/1";
26
27/// A file's path from the directory that holds `spec/`: a file of the tree, or one beside it.
28#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
29#[serde(transparent)]
30pub struct RootPath(String);
31
32impl RootPath {
33    /// A file of the tree.
34    #[must_use]
35    pub fn in_spec(path: &SpecPath) -> Self {
36        Self(format!("{SPEC}/{}", path.as_str()))
37    }
38
39    /// A file beside `spec/`, such as the sidecar.
40    #[must_use]
41    pub fn beside(name: &str) -> Self {
42        Self(name.to_owned())
43    }
44
45    pub(crate) fn as_str(&self) -> &str {
46        &self.0
47    }
48}
49
50/// What one save writes and removes.
51#[derive(Debug, Clone, Default, PartialEq, Eq)]
52pub struct Changes {
53    pub write: BTreeMap<RootPath, Vec<u8>>,
54    pub remove: BTreeSet<RootPath>,
55}
56
57impl Changes {
58    /// Writes each of the files of the tree and removes each of the paths under `spec/`.
59    #[must_use]
60    pub fn of_tree(write: Files, remove: &[SpecPath]) -> Self {
61        Self {
62            write: write
63                .into_iter()
64                .map(|(path, bytes)| (RootPath::in_spec(&path), bytes))
65                .collect(),
66            remove: remove.iter().map(RootPath::in_spec).collect(),
67        }
68    }
69
70    fn is_empty(&self) -> bool {
71        self.write.is_empty() && self.remove.is_empty()
72    }
73}
74
75#[derive(Debug, Serialize, Deserialize)]
76struct Plan {
77    format: String,
78    write: Vec<Staged>,
79    remove: Vec<RootPath>,
80}
81
82#[derive(Debug, Serialize, Deserialize)]
83struct Staged {
84    path: RootPath,
85    staged: String,
86}
87
88/// A save whose files are staged and whose plan is not yet committed.
89#[derive(Debug)]
90pub struct Prepared {
91    root: PathBuf,
92    plan: Plan,
93}
94
95/// A save whose plan is committed and whose files are not yet all in place.
96#[derive(Debug)]
97pub struct Committed(Prepared);
98
99/// Saves the changes: they are staged, committed and put in place. An empty set of changes
100/// writes nothing.
101///
102/// # Errors
103///
104/// A journal is already there, or a file cannot be written, renamed or removed. Before the
105/// commit every file is as it was; after it, [`recover`] finishes the save.
106pub fn save(root: &Path, changes: &Changes) -> Result<(), StoreError> {
107    if changes.is_empty() {
108        return Ok(());
109    }
110    prepare(root, changes)?.commit()?.finish()
111}
112
113/// Stages every file to write in the journal, each synced to disk.
114///
115/// # Errors
116///
117/// A journal is already there, or a staged file cannot be written.
118pub fn prepare(root: &Path, changes: &Changes) -> Result<Prepared, StoreError> {
119    let journal = root.join(JOURNAL);
120    if journal.exists() {
121        return Err(StoreError::Pending(journal));
122    }
123    let staged_dir = journal.join(STAGED);
124    fs::create_dir_all(&staged_dir).map_err(io_error(&staged_dir))?;
125    let mut write = Vec::with_capacity(changes.write.len());
126    for (index, (path, bytes)) in changes.write.iter().enumerate() {
127        let staged = index.to_string();
128        write_synced(&staged_dir.join(&staged), bytes)?;
129        write.push(Staged {
130            path: path.clone(),
131            staged,
132        });
133    }
134    sync_dir(&staged_dir)?;
135    Ok(Prepared {
136        root: root.to_owned(),
137        plan: Plan {
138            format: FORMAT.to_owned(),
139            write,
140            remove: changes.remove.iter().cloned().collect(),
141        },
142    })
143}
144
145impl Prepared {
146    /// Commits the save: the plan is written beside the staged files, synced, and renamed into
147    /// place. From here on the save happens, whether or not this process finishes it.
148    ///
149    /// # Errors
150    ///
151    /// The plan cannot be written or renamed; the save is then not committed.
152    pub fn commit(self) -> Result<Committed, StoreError> {
153        let journal = self.root.join(JOURNAL);
154        let staged = journal.join(PLAN_STAGED);
155        let bytes = serde_json::to_vec_pretty(&self.plan).expect("a plan serializes");
156        write_synced(&staged, &bytes)?;
157        let plan = journal.join(PLAN);
158        fs::rename(&staged, &plan).map_err(io_error(&plan))?;
159        sync_dir(&journal)?;
160        Ok(Committed(self))
161    }
162}
163
164impl Committed {
165    /// Puts every staged file in place, removes the files to remove, and deletes the journal.
166    ///
167    /// # Errors
168    ///
169    /// A file that cannot be renamed or removed; [`recover`] finishes the save later.
170    pub fn finish(self) -> Result<(), StoreError> {
171        finish(&self.0.root, &self.0.plan)
172    }
173}
174
175/// Finishes a committed save that was interrupted, or discards the staged files of one that was
176/// not committed. With no journal it does nothing.
177///
178/// # Errors
179///
180/// The plan cannot be read or names an unknown format, or a file cannot be renamed or removed.
181pub fn recover(root: &Path) -> Result<(), StoreError> {
182    let journal = root.join(JOURNAL);
183    if !journal.exists() {
184        return Ok(());
185    }
186    let plan_path = journal.join(PLAN);
187    match fs::read(&plan_path) {
188        Ok(bytes) => {
189            let plan: Plan = serde_json::from_slice(&bytes)
190                .map_err(|error| StoreError::Journal(plan_path.clone(), error.to_string()))?;
191            if plan.format != FORMAT {
192                return Err(StoreError::Journal(
193                    plan_path,
194                    format!("unknown format `{}`", plan.format),
195                ));
196            }
197            finish(root, &plan)
198        }
199        Err(error) if error.kind() == io::ErrorKind::NotFound => remove_journal(root),
200        Err(error) => Err(io_error(&plan_path)(error)),
201    }
202}
203
204/// Every step skips what an earlier, interrupted run already did, so finishing twice is safe.
205fn finish(root: &Path, plan: &Plan) -> Result<(), StoreError> {
206    let staged_dir = root.join(JOURNAL).join(STAGED);
207    let mut touched = BTreeSet::new();
208    for entry in &plan.write {
209        let staged = staged_dir.join(&entry.staged);
210        let target = root.join(entry.path.as_str());
211        if staged.exists() {
212            if let Some(parent) = target.parent() {
213                fs::create_dir_all(parent).map_err(io_error(parent))?;
214            }
215            fs::rename(&staged, &target).map_err(io_error(&target))?;
216        }
217        touched.extend(target.parent().map(Path::to_owned));
218    }
219    for path in &plan.remove {
220        let target = root.join(path.as_str());
221        match fs::remove_file(&target) {
222            Ok(()) => {}
223            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
224            Err(error) => return Err(io_error(&target)(error)),
225        }
226        remove_empty_parents(root, &target);
227        touched.extend(target.parent().map(Path::to_owned));
228    }
229    // A directory a later removal emptied is gone, and has nothing left to sync.
230    for dir in touched.iter().filter(|dir| dir.exists()) {
231        sync_dir(dir)?;
232    }
233    remove_journal(root)
234}
235
236/// Removes each directory above `target` that is left empty, up to but not including `spec/`.
237fn remove_empty_parents(root: &Path, target: &Path) {
238    let spec = root.join(SPEC);
239    let mut dir = target.parent();
240    while let Some(current) = dir {
241        if current == spec || !current.starts_with(&spec) || fs::remove_dir(current).is_err() {
242            break;
243        }
244        dir = current.parent();
245    }
246}
247
248/// Deletes the journal, and `.loft/` too when nothing else is in it.
249fn remove_journal(root: &Path) -> Result<(), StoreError> {
250    let journal = root.join(JOURNAL);
251    fs::remove_dir_all(&journal).map_err(io_error(&journal))?;
252    if let Some(parent) = journal.parent() {
253        let _ = fs::remove_dir(parent);
254    }
255    Ok(())
256}
257
258fn write_synced(path: &Path, bytes: &[u8]) -> Result<(), StoreError> {
259    let mut file = fs::File::create(path).map_err(io_error(path))?;
260    file.write_all(bytes).map_err(io_error(path))?;
261    file.sync_all().map_err(io_error(path))
262}
263
264/// Syncs a directory, so the renames into it survive a power loss. Only Unix can open a
265/// directory to sync it.
266fn sync_dir(dir: &Path) -> Result<(), StoreError> {
267    #[cfg(unix)]
268    fs::File::open(dir)
269        .and_then(|handle| handle.sync_all())
270        .map_err(io_error(dir))?;
271    #[cfg(not(unix))]
272    let _ = dir;
273    Ok(())
274}
loftspec_store/journal.rs

loftspec_store/
journal.rs
