alembic_engine/
journal.rs

1//! Keep track of successfully applied ops to enable resume after an error.
2//!
3//! When resuming, the journal must match the previous run's non-delete op sequence (including op hashes).
4
5use crate::{BackendId, Op};
6use alembic_core::{TypeName, Uid};
7use anyhow::{anyhow, Result};
8use serde::{Deserialize, Serialize};
9use std::fs;
10use std::fs::{File, OpenOptions};
11use std::io::{Read, Seek, Write};
12use std::path::{Path, PathBuf};
13use tempfile::NamedTempFile;
14
15#[derive(Debug, Serialize, Deserialize)]
16pub struct Journal {
17    #[serde(skip)]
18    file: Option<(File, PathBuf)>,
19    ops: Vec<OpWithMeta>,
20}
21
22impl Journal {
23    pub fn stable_file_name(directory: &Path, adapter_name: &str, ops: &[Op]) -> PathBuf {
24        let hash = crate::types::stable_json_hash(&ops);
25        let file_name: PathBuf = format!("{}_journal_{}.yaml", adapter_name, hash).into();
26        directory.join(file_name)
27    }
28
29    /// tries to load a Journal from `file_path`, otherwise creates a new one.
30    /// in either case, the new Journal instance will be backed by the file at `file_path`.
31    /// delete ops will not be saved in the journal.
32    pub fn load_or_create(directory: &Path, adapter_name: &str, ops: &[Op]) -> Result<Self> {
33        // apply writes the journal before any state save, so `directory` (e.g. `.alembic/`)
34        // may not exist yet on a fresh checkout.
35        fs::create_dir_all(directory)?;
36        let file_name = Self::stable_file_name(directory, adapter_name, ops);
37        if fs::metadata(&file_name).is_ok() {
38            Self::new_from_existing_file(directory, adapter_name, ops)
39        } else {
40            Self::new_with_file(directory, adapter_name, ops)
41        }
42    }
43
44    /// loads a journal from the file with `file_path` and sets its backing file to that file
45    fn new_from_existing_file(
46        directory: &Path,
47        adapter_name: &str,
48        expected_ops: &[Op],
49    ) -> Result<Self> {
50        let file_name = Self::stable_file_name(directory, adapter_name, expected_ops);
51
52        let mut file = OpenOptions::new()
53            .read(true)
54            .write(true)
55            .create(false)
56            .append(false)
57            .open(&file_name)?;
58        let mut contents = String::new();
59        file.read_to_string(&mut contents)?;
60
61        let mut journal: Journal = serde_yaml::from_str(&contents)?;
62
63        let journal_keys = journal
64            .ops
65            .iter()
66            .map(|op_with_meta| {
67                (
68                    op_with_meta.op_uid,
69                    &op_with_meta.op_typename,
70                    op_with_meta.op_hash,
71                )
72            })
73            .collect::<Vec<(Uid, &TypeName, u64)>>();
74
75        let expected_keys = expected_ops
76            .iter()
77            .filter(|op| !matches!(op, Op::Delete { .. }))
78            .map(|op| (op.uid(), op.type_name(), op.hashed()))
79            .collect::<Vec<(Uid, &TypeName, u64)>>();
80        if journal_keys != expected_keys {
81            return Err(anyhow!(
82                "the ops in the loaded journal file `{}` don't match the expected ops",
83                file_name.display()
84            ));
85        }
86
87        journal.file = Some((file, file_name));
88        Ok(journal)
89    }
90
91    /// creates a journal with a new backing file
92    fn new_with_file(directory: &Path, adapter_name: &str, ops: &[Op]) -> Result<Self> {
93        let file_name = Self::stable_file_name(directory, adapter_name, ops);
94        let mut journal = Self::new_ephemeral(ops);
95
96        // create and write to the file to check that it works before applying any ops
97        let mut file = OpenOptions::new()
98            .read(true)
99            .write(true)
100            .create(true)
101            .truncate(true)
102            .append(false)
103            .open(&file_name)?;
104        file.set_len(0)?;
105        file.rewind()?;
106
107        journal.file = Some((file, file_name));
108        journal.save()?;
109
110        Ok(journal)
111    }
112
113    /// creates a journal without a backing file set
114    pub fn new_ephemeral(ops: &[Op]) -> Self {
115        Self {
116            file: None,
117            ops: ops
118                .iter()
119                .filter(|op| !matches!(op, Op::Delete { .. }))
120                .map(OpWithMeta::new)
121                .collect(),
122        }
123    }
124
125    pub fn done_ops(&self) -> Vec<(Uid, TypeName, u64)> {
126        self.ops
127            .iter()
128            .filter(|owm| owm.done)
129            .map(|owm| (owm.op_uid, owm.op_typename.clone(), owm.op_hash))
130            .collect()
131    }
132
133    pub fn done_ops_count(&self) -> usize {
134        self.ops.iter().filter(|op| op.done).count()
135    }
136
137    pub fn is_completed(&self) -> bool {
138        self.ops.iter().all(|op| op.done)
139    }
140
141    /// will mark the first op that is not done (will fail if there's no such op).
142    /// uses a linear search from the start.
143    pub fn mark_op_as_done(&mut self, op: &Op) -> Result<()> {
144        let op_hash = op.hashed();
145        let op_uid = op.uid();
146        let op_typename = op.type_name();
147
148        let Some(op_index) = self.ops.iter().position(|op| {
149            !op.done
150                && op.op_hash == op_hash
151                && op.op_uid == op_uid
152                && &op.op_typename == op_typename
153        }) else {
154            return Err(anyhow!(
155                "no matching op found in journal, can't mark any as done"
156            ));
157        };
158
159        // index comes from call to `position` above, so it must be in range
160        self.ops[op_index].done = true;
161
162        Ok(())
163    }
164
165    pub fn save(&mut self) -> Result<()> {
166        let str = serde_yaml::to_string(self)?;
167
168        let (_, path) = self
169            .file
170            .as_ref()
171            .ok_or_else(|| anyhow!("can't save journal because it's missing a backing file"))?;
172
173        let path = path.clone();
174        let dir = path
175            .parent()
176            .ok_or_else(|| anyhow!("file path has no parent directory"))?;
177
178        let mut temp_file = NamedTempFile::new_in(dir)?;
179        temp_file.write_all(str.as_bytes())?;
180        temp_file.as_file().sync_all()?; // fsync data + metadata before it can become visible
181        temp_file.persist(&path)?;
182        File::open(dir)?.sync_all()?;
183        let new_file = OpenOptions::new().read(true).write(true).open(&path)?;
184        self.file = Some((new_file, path));
185
186        Ok(())
187    }
188
189    pub fn delete_backing_file(&mut self) -> Result<()> {
190        if let Some((file, file_path)) = self.file.take() {
191            drop(file);
192            fs::remove_file(file_path)?;
193        }
194        Ok(())
195    }
196}
197
198impl Drop for Journal {
199    fn drop(&mut self) {
200        if let Some((file, _)) = self.file.take() {
201            let _ = file.sync_all();
202        }
203    }
204}
205
206// we're only storing the uid and typename for the Op to keep this struct small and readable
207#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
208struct OpWithMeta {
209    op_uid: Uid,
210    op_typename: TypeName,
211    op_hash: u64,
212    done: bool,
213    backend_id: Option<BackendId>, // FIXME: not sure if/when this is needed
214}
215
216impl OpWithMeta {
217    fn new(op: &Op) -> Self {
218        OpWithMeta {
219            op_uid: op.uid(),
220            op_typename: op.type_name().clone(),
221            op_hash: op.hashed(),
222            done: false,
223            backend_id: None,
224        }
225    }
226}
227
228#[cfg(test)]
229mod tests {
230    use super::*;
231    use alembic_core::{Object, TypeName};
232    use tempfile::tempdir;
233
234    fn test_ops() -> Vec<Op> {
235        vec![
236            Op::Create {
237                uid: Uid::from_u128(1),
238                type_name: TypeName::new("dcim.device"),
239                desired: Object {
240                    uid: Uid::from_u128(1),
241                    type_name: TypeName::new("dcim.device"),
242                    key: Default::default(),
243                    attrs: Default::default(),
244                    source: None,
245                },
246            },
247            Op::Create {
248                uid: Uid::from_u128(2),
249                type_name: TypeName::new("dcim.device"),
250                desired: Object {
251                    uid: Uid::from_u128(2),
252                    type_name: TypeName::new("dcim.device"),
253                    key: Default::default(),
254                    attrs: Default::default(),
255                    source: None,
256                },
257            },
258            Op::Update {
259                uid: Uid::from_u128(3),
260                type_name: TypeName::new("dcim.site"),
261                desired: Object {
262                    uid: Uid::from_u128(3),
263                    type_name: TypeName::new("dcim.site"),
264                    key: Default::default(),
265                    attrs: Default::default(),
266                    source: None,
267                },
268                changes: vec![],
269                backend_id: None,
270            },
271        ]
272    }
273
274    #[test]
275    fn journal_identity_is_stable_across_toolchains() {
276        // pinned constants: the journal file name and per-op hashes are persisted
277        // to disk and compared on resume, so they must not depend on the rust or
278        // dependency versions. if this test breaks, cross-version resume broke.
279        let ops = test_ops();
280        assert_eq!(ops[0].hashed(), 2629757075790778505);
281        assert_eq!(
282            Journal::stable_file_name(Path::new("/tmp"), "test", &ops),
283            PathBuf::from("/tmp/test_journal_5133211470659736648.yaml")
284        );
285    }
286
287    #[test]
288    fn save_and_load_journal() {
289        let dir = tempdir().unwrap();
290        let ops = test_ops();
291        {
292            let mut journal = Journal::new_with_file(dir.path(), "test", &ops).unwrap();
293            journal.save().unwrap();
294            drop(journal);
295        }
296        {
297            let journal = Journal::new_from_existing_file(dir.path(), "test", &ops).unwrap();
298            assert_eq!(
299                journal
300                    .ops
301                    .iter()
302                    .map(|owm| (owm.op_uid, &owm.op_typename))
303                    .collect::<Vec<(Uid, &TypeName)>>(),
304                ops.iter()
305                    .map(|op| (op.uid(), op.type_name()))
306                    .collect::<Vec<(Uid, &TypeName)>>()
307            );
308        }
309    }
310
311    #[test]
312    fn load_and_save_existing_journal() {
313        let dir = tempdir().unwrap();
314        let ops = test_ops();
315        {
316            let mut journal = Journal::new_with_file(dir.path(), "test", &ops).unwrap();
317            journal.save().unwrap();
318        }
319        {
320            let mut journal = Journal::new_from_existing_file(dir.path(), "test", &ops).unwrap();
321            journal.save().unwrap();
322            assert_eq!(journal.ops.len(), 3);
323        }
324    }
325
326    #[test]
327    fn mark_ops_as_done() {
328        let ops = test_ops();
329        let mut journal = Journal::new_with_file(tempdir().unwrap().path(), "test", &ops).unwrap();
330        journal.mark_op_as_done(&ops[0]).unwrap();
331        assert!(!journal.is_completed());
332        journal.mark_op_as_done(&ops[1]).unwrap();
333        assert!(!journal.is_completed());
334        journal.mark_op_as_done(&ops[2]).unwrap();
335        assert!(journal.is_completed());
336    }
337
338    #[test]
339    fn mark_ops_as_done_backwards() {
340        let ops = test_ops();
341        let mut journal = Journal::new_with_file(tempdir().unwrap().path(), "test", &ops).unwrap();
342        journal.mark_op_as_done(&ops[2]).unwrap();
343        assert!(!journal.is_completed());
344        journal.mark_op_as_done(&ops[1]).unwrap();
345        assert!(!journal.is_completed());
346        journal.mark_op_as_done(&ops[0]).unwrap();
347        assert!(journal.is_completed());
348    }
349
350    #[test]
351    fn mark_invalid_op_as_done() {
352        let ops = test_ops();
353        let mut journal = Journal::new_with_file(tempdir().unwrap().path(), "test", &ops).unwrap();
354        journal
355            .mark_op_as_done(&Op::Create {
356                uid: Uid::from_u128(999),
357                type_name: TypeName::new("dcim.site"),
358                desired: Object {
359                    uid: Uid::from_u128(999),
360                    type_name: TypeName::new("dcim.site"),
361                    key: Default::default(),
362                    attrs: Default::default(),
363                    source: None,
364                },
365            })
366            .expect_err("should fail");
367        assert!(!journal.is_completed());
368    }
369
370    #[test]
371    fn mark_same_op_as_done_twice() {
372        let ops = test_ops();
373        let mut journal = Journal::new_with_file(tempdir().unwrap().path(), "test", &ops).unwrap();
374        journal.mark_op_as_done(&ops[1]).unwrap();
375        journal.mark_op_as_done(&ops[1]).expect_err("should fail");
376    }
377
378    #[test]
379    fn delete_backing_file() {
380        let dir = tempdir().unwrap();
381        let ops = test_ops();
382        let mut journal = Journal::new_with_file(dir.path(), "test", &ops).unwrap();
383        journal.save().unwrap();
384        let file_path = Journal::stable_file_name(dir.path(), "test", &ops);
385        assert!(file_path.exists());
386        journal.delete_backing_file().unwrap();
387        assert!(!file_path.exists());
388    }
389}
alembic_engine/journal.rs

alembic_engine/
journal.rs
